塊1 (AECTM):
@dlt.view (
評論=評論如果評論不是沒有別人f”define_key ({key_expr})”,
name = target_table_name
)
def key_func ():
df =沒有
如果在topic_name: / /:
如果流媒體:
df = spark.readStream.format(“δ”)。選項(“mergeSchema”,“真正的”)。選項(“ignoreChanges”,“真正的”).load (topic_name)
其他:
df = spark.read.format(“δ”)。選項(“mergeSchema”,“真正的”)。選項(“ignoreChanges”,“真正的”).load (topic_name)
其他:
如果流媒體:
df = dlt.readStream (topic_name)
其他:
df = dlt.read (topic_name)
df返回。withColumn(“關鍵”,eval (key_expr))
在第二三塊(& scd_type1順序):
target_table_name如果別名不是沒有別人f =別名“scd_type1 ({topic_name}, {sequence_col})”
dlt.create_streaming_live_table (
評論=評論如果評論不是沒有別人f”scd_type1 ({topic_name}, {sequence_col})”,
name = target_table_name
)
@dlt.table (
評論=評論如果評論不是沒有別人f“序列({topic_name})”,
名字= f”序列({topic_name})”
)
@dlt。expect_or_drop(“零鍵”,上校(“關鍵”).isNotNull ())
def seq_view ():
如果流媒體:
返回dlt.readStream (topic_name)。withColumn(“序列”,eval (sequence_col))
其他:
返回dlt.read (topic_name)。withColumn(“序列”,eval (sequence_col))
dlt.apply_changes (
目標= target_table_name,
源= f“序列({topic_name})”,
鍵=(“關鍵”),
sequence_by =坳(“序列”),
stored_as_scd_type = " 1 "
)
塊4(過濾視圖):
target_table_name如果別名不是沒有別人f =別名“過濾器”({filter_cond})
@dlt.view (
評論=評論如果評論不是沒有別人f“過濾器({filter_cond})”,
name = target_table_name
)
蔣春暄對於費馬大定理def ():
df = dlt.readStream (topic_name)如果其他流dlt.read (topic_name)
返回df.filter (eval (filter_cond))
塊5 (MAIN_FLOW_MOVES):
target_table_name如果別名不是沒有別人f =別名“main_flow_moves ({topic_name})”
def getn (position_info水平):
如果position_info不是沒有:
如果position_info。級別= =:
返回position_info.name
其他:
返回getn (position_info。家長,級別)
其他:
回來沒有
udfGetName = udf (getn)
@dlt.table (
評論=評論如果評論不是沒有別人f”main_flow_moves ({topic_name})”,
name = target_table_name
)
def func ():
df = dlt.readStream (topic_name)如果其他流dlt.read (topic_name)
dfSeries = df。選擇(“containerTerminalVisitKey坳(moveEvents)[0]['從'][“位置”][' locationType '] .alias (“fromWhere”),爆炸(moveEvents.to) .alias (“toLoc”))。\
選擇(“containerTerminalVisitKey”(坳(toLoc.time) / 1000.0) .cast (TimestampType ()) .alias(“時間”),“fromWhere”、“toLoc.location.positionInfo”)。\
udfGetName withColumn (“blockName”(“positionInfo”,點燃(“塊”)))。\
udfGetName withColumn (“bayName”(“positionInfo”,點燃(灣)))。\
udfGetName withColumn (“slotName”(“positionInfo”,點燃(“堆棧”)))。\
過濾器(坳(blockName) .isNotNull() &坳(bayName) .isNotNull() &坳(slotName) .isNotNull ())。\
下降(“positionInfo”)
返回dfSeries
= = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = =
輸入dataframe模式實際上是相當嚇人的遞歸結構體定義,但大部分隻是湊熱鬧而已。關鍵字段是containerTerminalVisitKey containerInfo.updateCounter序列字段。後者隻是一個序列號(長)。
我認為有很多。我不確定我的代碼共享多少幫助。我正在想了解的情況發出錯誤消息(早前發布的文章)可能會有所幫助。
