加入一個大的數據量會導致磁盤空間誤差,如何接待?

[Erik_L]

我想做的

df = #所有id的沒有一個是有效的id在IDs中:#來自不同數據源的數據得到的部分df_1 = spark.read.parquet (url_for_id) df_2 = spark.read.parquet (url_for_id)…#結合部分基於時間和他們共同列id df_join = df_1。加入(= idx_columns df_2, =內部). join () ....withColumn (“id”,點燃(id)) #加入所有id到一個大表,如果df = =沒有:df =其他df_join: df.union (df_join) #現在寫出來與df.writeTo新的工作表(表).createOrReplace ()

每個ID都有100 GB的數據分解成0.5 - 1.5 GB的塊。聯盟的總大小表將在7和10 TBs。

發生了什麼事

引起的:java . util . concurrent。ExecutionException: . lang。RuntimeException:內部:/ local_disk0 /火花-…/執行器-…/ dbio_cache_root_…/ unified_cache -…/…/…/…:設備上沒有剩餘空間

似乎工會不聰明,因為當我每個ID執行這個級別,它將正常工作。

我需要幫助

什麼是最好的方式來處理這些大數據連接?我怎麼能讀和寫數據是否有效?

例如:

• 我應該加入每個表就立即運行“.writeTo(表).append ()”?會引發解決這一事實我這樣做10次在同一時間嗎?
• 檢查點可以幫助這個過程嗎?
• 有更好的辦法,聯盟嗎?
• 困難的部分是,ID是在對象的路徑,所以我閱讀每條路徑和添加亮()。我試著使用文件路徑和正則表達式,但運行速度較慢,因為它行而不是id。因此,使用流可能不會工作。
• 數據的時間戳,但火花不是保持本地集群的數據。我應該忽視磚的卡爾時間聚類優化和力的一些選項?z - index將永遠如此規模的數據和數據已經在本地集群,所以我想一直這樣在攝入。

請幫助。

[匿名]

@Erik路易:

有幾個策略,您可以使用它們來處理大型加入這樣的火花:

1. 用廣播加入:如果你的一個dataframes相對較小(小於10 - 20 GB),您可以使用一個廣播加入避免交換數據。可以使用廣播加入dataframe遠小於另一個時,火花可以容納整個小dataframe到內存中每個節點上。您可以使用()函數顯式地播出dataframe播出。
2. 增加調整分區大小:當您執行加入火花,節點之間的數據打亂。默認情況下,使用200火花洗牌分區,但是你可以增加這個數字減少每個分區的數據重組。你可以設置spark.sql.shuffle。分區配置參數更大的值(例如,1000)增加洗牌分區的數量。
3. 用桶加入:如果你的數據已經分區或集群通過加入的一個關鍵,您將使用,您可以使用一個桶加入避免交換數據。桶加入要求dataframes都鬥通過相同的鍵,按相同的鍵,那麼您可能需要重組相應數據。您可以使用bucketBy()和sortBy()函數dataframe桶和排序。
4. 使用檢查點機製:如果你在加入耗盡內存,你可以使用一個檢查點機製中間結果寫入磁盤。這可以幫助防止大型dataframes內存不足錯誤和提高性能。您可以使用檢查點()函數dataframe檢查點,並可以設置spark.sql。autoBroadcastJoinThreshold配置參數更大的值觸發自動對大量dataframes檢查點。

在閱讀和寫作方麵的數據,您可能需要考慮使用分布式文件係統(比如HDFS或S3存儲您的數據,它可以幫助可擴展性和容錯性。您可以使用

讀()和寫()函數在火花從這些文件係統讀取和寫入數據。

總的來說,最好的策略來處理大型連接將取決於你的數據和用例的細節。你可能需要嚐試不同的方法和配置需要找到最優解。

[Erik_L]

謝謝你的全麵的反應——我看看其中的一些。額外的上下文,這裏有一些限製數據:

所有的表都應該_mostly_相同的尺寸和相同的密鑰生成過程,生成時間序列數據(也就是說,拚花表的數據源是一係列在S3中按時間排序,每個時間戳將_usually_存在於所有N表,但不總是這樣)。攝入之前,這些表是按時間戳排序,但火花打亂的攝取,因為我運行一個unix時間戳一個ISO的時間戳(我真的想避免這種情況,如果可能的話,但是我不知道除了“bucketBy”和“sortBy”之前建議)。

不幸的是,我沒有控製的方式生成這些表,但在未來我意願生成這些表預連接而不是分開,因為沒有價值讓他們分開。