嗨,專家,
我需要攝取數據從現有自己的三角洲湖三角洲路徑。數據流圖所示:
我已經嚐試以下代碼:
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個問題:
最好的問候,
Vu