流三角洲生活表,如果我重新運行的管道,它將新數據附加到當前表嗎?

[麥當娜]

你好,

我有一個問題關於DLT表。

假設我有一個流DLT管道從青銅表讀取數據並應用轉換數據。

管道模式觸發。

如果我重新運行管道,將新數據附加到當前表嗎?如果是這樣,如何確保沒有重複生成的表嗎?

在使用結構化流的情況下,方法有一個選項“writeStreamoutputMode”來控製如何寫數據。但是,DLT管道沒有“writeStream”方法。所以,我怎麼能控製寫入數據到DLT表嗎?

[匿名 · Accepted solution]

@Mohammad劍:

在磚三角洲湖(DLT)管道,當你重新運行管道在“附加”模式下,新數據將添加到現有表。三角洲湖提供內置支持處理重複通過“插入”功能。您可以使用“合並”命令合並新數據與現有的數據表中根據特定的條件。這可以通過使用一個主鍵或惟一標識符確保副本不添加到表中。

控製寫數據到DLT管道DLT表,您可以使用三角洲湖API來編寫數據表。這可以通過使用PySpark deltaTable對象,它提供了一個接口,用於讀取或寫入三角洲表。寫數據到三角洲表時,您可以使用模式參數來控製數據是如何寫的。模式參數值可以像“追加”,“覆蓋”和“忽視”,等等。以確保副本不添加到表,您可以使用“插入”模式,由三角洲湖。

[麥當娜]

由於@Suteja卡努裏人

如果有任何樣本筆記本如何控製寫入數據到DLT表DLT管道使用三角洲湖API,你能和我分享它嗎?

謝謝。

[匿名]

@Mohammad軍刀:你可以看看https://www.dbdemos.ai/

也給你一些代碼

#從三角洲從pyspark.sql進口DeltaTable進口必要的庫。功能導入* # Define三角洲湖表路徑table_path =“/ mnt /δ/ my_table”#數據加載到火花DataFrame df = spark.read.format (csv)。選項(“頭”,“真正的”).load (“/ mnt / my_data.csv”) #過濾數據隻包含行一定值df_filtered = df.filter(坳(my_column) = =“my_value”) #創建DeltaTable對象表delta_table = DeltaTable。forPath(火花,table_path) #檢查表是否存在,如果不如果不是DeltaTable創建它。isDeltaTable(火花,table_path): delta_table。創建(df_filtered。模式,partitionBy = " my_column ") #將過濾後的數據插入到表delta_table.alias (“t”)。合並(df_filtered.alias (“s”),“t。my_column =。my_column”) .whenNotMatchedInsertAll () . execute ()

在這個例子中,我們首先將一些數據加載到火花DataFrame和過濾僅包含行有一定的價值。然後我們創建一個DeltaTable對象DLT表在指定路徑和檢查它是否存在。如果表不存在,我們創建的模式過濾DataFrame和分區列。最後,我們使用DeltaTable合並()函數來將過濾後的數據插入到表中。merge()函數執行插入操作,更新行匹配給定條件和插入行不。在本例中,我們使用my_column列合並條件,這就意味著如果一行具有相同的價值my_column表中已經存在,它將被更新值過濾DataFrame,如果它不存在,將插入一個新行。

[Valentin1]

你能改變流管道的outputMode三角洲住表嗎?(我想恰克更新模式中使用“applyInPandasWithState”outputMode“更新”。有什麼區別“追加”三角洲生活表,因為它似乎工作嗎?)