添加重複數據刪除方法引發流

[patojo94]

大家好,我有一些麻煩添加重複數據刪除步驟已經運行的文件流。我想添加的代碼是這個:

df = df。withWatermark (“arrival_time”、“20分鍾”)\ .dropDuplicates ([“event_id”、“arrival_time”])

然而,我得到以下錯誤。

引起的:. lang。HDFSStateStoreProvider IllegalStateException:錯誤閱讀流狀態文件[id = (op = 0 = 101), dir = dbfs: / mnt /檢查點/ silver_events /州/ 0/101):dbfs: / mnt /檢查點/ silver_events /州/ 0/101/1.delta並不存在。如果流的工作重新開始一個新的或更新的狀態操作,請創建一個新的檢查點位置或清除現有檢查點位置。

錯過兩個問題是:

1. 為什麼我得到這個錯誤,這是什麼意思?
2. 真的可以刪除一個流檢查點和重啟流時不會重複的數據?

謝謝你！

[Kaniz · Accepted solution]

嗨@patricio東城,你可以減少重複記錄在事件數據流使用一個惟一的標識符。這是完全相同的重複數據刪除靜態使用一個惟一的標識符列。必要的查詢將存儲的數據量從以前的記錄,這樣它可以過濾重複的記錄。類似於聚合,可以使用重複數據刪除有或沒有水印。

• 有水印——如果有一個上限多晚複製記錄可以到達,你可以定義一個事件時間列上水印和減少重複使用guid和事件時間列。查詢將使用水印從過去的記錄,刪除舊的狀態數據預計不會得到任何副本了。這個界限的狀態查詢必須維護。

• 沒有水印——因為沒有界限複製記錄可能到達的時候,查詢存儲數據從所有過去記錄的狀態。

https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html streaming-deduplica……

解決方案1。

• 你的檢查點位置是什麼?這通常可以因為你使用本地文件係統存儲檢查點。確保你設置“檢查點位置”選項,它指向一個分布式文件係統(比如HDFS)所有節點可以訪問。
• https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html streaming-deduplica……

解決方案2。

• 你不應該分享檢查點目錄不同流查詢。
• 使用一個新的檢查點為每一個新的流媒體查詢目錄。
• https://kb.www.eheci.com/delta/delete-checkpoint-restart.html

[Kaniz]

嗨@patricio故事,我們還沒有收到你最後的回應我,我被檢查回看到如果你有一個決議。如果你有任何解決方案,請與社區分享,因為它可以幫助別人。否則,我們將與更多的細節和試圖幫助回應。