結構化流從一個覆蓋三角洲路徑

[Vu_QuangNguyen]

嗨,專家,

我需要攝取數據從現有自己的三角洲湖三角洲路徑。

數據流圖所示:

• 數據團隊讀取數據庫表的快照和三角洲路徑覆蓋。這是每天做了很多次,但並不是每天固定的時間表。

我需要每一個數據流改變自己對下遊的三角洲湖消費。與源三角洲湖基本相同,但與增加日誌和數據保留時間,使旅行三年。

我已經嚐試以下代碼:

def overwrite_microbatch (microdf batchId): microdf.write.format(“δ”).mode(“覆蓋”).save (sink_path)(火花。readStream .format(“δ”).option (“ignoreChanges”,“真正的”).load (source_path) .writeStream .foreachBatch (overwrite_microbatch) .option (“checkpointLocation”, checkpoint_path) .start ())

(.writeStream.format(“δ”).outputMode(“追加”)不工作,因為“追加”模式導致重複和writeStream不支持“覆蓋”模式)。

工作,但是我遇到了2個問題:

1. 水槽路徑不是存儲優化,我。e每個版本存儲.snappy全表快照。鑲木地板文件而不是隻有增量更改。
2. 如果我流的工作無法使用一個或多個版本,然後下一個microbatch包含concat 2 +版本尚未消耗。這導致水槽重複路徑。

這個場景的正確的方法應該是怎樣的?任何想法非常感謝。謝謝!

最好的問候,

Vu