三角洲表不能進入信息從以前的表的管道

[anasse]

你好

我的新磚使用。我想創造一些複雜的轉換在三角洲表管道。

我有一些表,在流媒體模式來收集數據從一個S3然後銀層開始轉換,但在我看來像金層開始在年底前銀所以我所有的黃金表是空的,我需要刷新他們獨立……

有句話我失蹤在這個過程嗎?

謝謝你的幫助

[Kaniz]

嗨@Anasse Berahab,你可能會經曆一個同步問題金銀層三角洲湖之間的管道。為了解決這個問題,您可以使用觸發器和awaitTermination選項來控製流的執行查詢。

這是一個大綱來幫助你正確地設置管道:

1. 讀取數據從源(例如,S3)並將它寫入銅層。
2. 轉換數據的銅層,把它寫在銀層。
3. 轉換數據的銀層,把它寫在金層。

確保金層轉換開始隻有銀層轉換完成後,您可以按照以下步驟:

1. 開始為銀層流式查詢轉換使用寫流和指定觸發器來控製處理的頻率將被執行。

例如:

silver_streaming_query = (bronze_df .transform (silver_transform) .writeStream .format .outputMode(“δ”)(“追加”).option (“checkpointLocation”、“/道路/ /銀/檢查站”).trigger (processingTime =“5分鍾”).start(路徑“/ / /銀/輸出”))

這裏,silver_transform是一個函數,定義了轉換邏輯從青銅銀。替換/道路/ /銀/檢查站和/道路/ /銀/輸出合適的路徑為你的用例。

1. 啟動流查詢使用writeStream金層轉換:

gold_streaming_query = (silver_df .transform (gold_transform) .writeStream .format .outputMode(“δ”)(“追加”).option (“checkpointLocation”、“/道路/ /金/檢查站”).trigger (processingTime = 10分鍾).start(“/道路/ /金/輸出”))

這裏,gold_transform是一個函數,定義了轉換邏輯的白銀和黃金。替換/道路/ /金/檢查站和/道路/ /金/輸出與相應的路徑為你的用例。

1. 使用awaitTermination等待金銀流查詢來完成:

gold_streaming_query.awaitTermination silver_streaming_query.awaitTermination () ()

通過設置和使用awaitTermination觸發選項,您可以控製流的順序和頻率轉換。通過這種方式,您可以確保金層轉換後才開始銀層轉換完成。

如果您仍然遇到問題,考慮檢查日誌任何錯誤或警告消息。

此外,確保您的轉換邏輯是正確的,你沒有過濾掉所有的數據在轉換。