如何從一個變量隻最近記錄的表數流

[tom_shaffner]

短的版本我需要一種方法來從一個變量隻最近記錄的表數流。這是一個相對簡單的問題在sql或python熊貓(group by和最新的),但在流,我不停的按塊。我可以流或合並外的全部曆史和創建一個視圖隻是最近,但我預測查詢這些結果表非常定期和重新計算他們每天或每一個查詢會浪費快。坦率地說,在這一點上我隻是建立一個傳統數據庫有點違背磚的目的。

長版本:我們的許多數據feed來作為單獨的實體,而我需要合並成一個單一的“當前”表。這意味著我要把最近的記錄,通過ID,從源表(變量數,可以是兩個以上)到一個最終表,所有使用流和可概括的方式進行重用。

首先,如果我試著通過SQL臨時視圖合並方法討論了許多磚學院視頻(這讓我創建一個有點類似SQL的解決方案)我得到這個:我不能使用Append模式,然而隨著這個加入的目的是更新新記錄時相關(例如,如果一個事件是一個更新、輸出表中記錄應該被取代)。可能,我可以使用完整模式相反,但這將重建整個目標表每次我相信,意味著我失去所有的流效率。

我也試過的方法,我做了一個stream-stream加入pyspark直接從多個源表和加入的靜態子查詢的結果。這有相當複雜的我不得不加入的靜態dataframe與多個流dataframes子查詢,找出所有的水印,並加入那些dataframes。最後我還是打上麵的錯誤。

我也有另一種方法它把東西分解成多個筆記本分裂插入事件(附加)更新事件(插入)但是我語法錯誤也在這些,這種方法需要一個任務鏈,可以得到複雜而減慢速度的結果。也許這是唯一可行的方法,但調試這些個人錯誤,特別是當upsert應該取代隻有更新的時間戳,被證明是複雜的。

在這,我突然想到我應該問這裏有一個建議的解決方案。

[Hakon_Amdal]

你試過存儲所有三角洲表合並成[1]嗎?您可以指定一個條件“當匹配”,這樣你隻插入是否更新的時間戳。

[1]https://docs.m.eheci.com/spark/latest/spark-sql/language-manual/delta-merge-into.html

[tom_shaffner]

是的,“附加”模式是做pyspark Upsert /合並,這是一樣的SQL合並成我相信;這就是給了我上麵的錯誤。

我有另一個版本做了合並成單個數據源,但我跟磚和他們建議我這樣做需要從每個源順序合並避免衝突寫道,和一個錯誤會導致失敗在他們所有人。

[tom_shaffner]

@Hakon Amdal需要澄清的是,我認為你的建議實際上是唯一可行的方法。即每個數據源的插入,按順序。

這是唯一的方法似乎是一個火花限製;答案在apache火花——結構化流提取每個id - Stack Overflow的最新值討論了。網上有多個解決方案討論聯合方式,但似乎沒有一個回答者的工作,也不會對我而言,這意味著需要順序而不是共同完成的。

我可能仍然開關的方法,但是之前我想更好地理解為什麼磚工程師我和推薦的反對和選擇是什麼。工程師的建議似乎表明要更noSQL-like更新樣式表和處理結果通過聚合函數;通過了解,我仍然工作。