磚

到目前為止我沒有任何foreachbatch代碼。我在執行dedup整個數據來自事件中心

我能看到丟失的記錄在事件中心。我應該使用forEachBatch在我的代碼。?或任何其他建議。

我基於我的主鍵刪除重複的多個記錄來自事件的中心一個主鍵和我想要最新的一個也我做訂單op_ts列以便我得到最新的記錄。明顯的可以創造…

使用以下代碼:參看= {}df = spark.readStream.format (eventhubs) .options (* * conf) .load () dataDF = df.select(坳(“身體”).cast(“字符串”))data = dataDF.select (json_tuple(坳(“身體”),“表”、“op_type”、“記錄”、“op_ts”)) \ .toDF(“表”、“op_……

我找不到那些丟失的記錄數據湖