val df1= spark.read.format("delta").table("...100K_rows...")
val xform = udf ( (message: String) => {
// abstract transformation, ideally this comes from a .jar library
// such as: (abstract) https://github.com/cosmycx/scala_transformer
val t0 = System.currentTimeMillis
Thread.sleep(5)
System.currentTimeMillis - t0
}) // .xformText
spark.udf.register("xform", xform)
val df2= df1.withColumn("xformResult", xform($"SomeText"))
df2.write.format("delta")
.mode(SaveMode.Overwrite)
.saveAsTable("...")
如何讓它運行得更快？
我嘗試了什么：
- 增加databricks spark集群節點大小：DS3_v2 14GB 4 cores vs. DS5_v2 56GB 16 cores
- 增加資料塊火花集群工人數量，驅動程式加：3、5和10（相同的速度！？）
- 改變：spark.conf.set("spark.sql.shuffle.partitions", "auto") 或不同的值
結果始終在此范圍內：10K 行為 1 分鐘，100K 行為 8 分鐘，無論更改如何。
100K 的理想結果不到 1 分鐘。這甚至可以在databricks spark中實作嗎？如果重要，它會在 Azure 中運行。
我錯過了什么，其他要考慮的事情，試試？謝謝。
uj5u.com熱心網友回復：
首先，這可能不是 spark 的用例。資料集很小，火花優化器甚至不知道如何處理它。
發生的情況是檔案很小并且由單個任務讀取，spark 將轉換識別為簡單并在讀取資料時應用它 - 因此您按順序處理事物并等待 Thread.sleep() 的總 8 分鐘
最好的想法可能是不要為此使用 spark - 但是如果您確實嘗試讓 spark 進行 shuffle（排序或其他）并確保您的代碼在 shuffle 之后運行，那么您將擁有 x（默認為 200）個磁區并且最多可以運行 x 個不同的任務（但同樣 - 我不會為此使用 Spark）
uj5u.com熱心網友回復：
由于 df1 DataFrame 只有一個磁區，因此轉換是按順序進行的，而不是并發的。重新磁區初始 DataFrame，然后運行轉換，新磁區 DataFrame 將速度和性能顯著提高至少 10 倍。
println(df1.rdd.getNumPartitions) // 1
val df2 = df1.repartition(20) // run transform on df2 (100K rows in 35 sec.)
轉載請註明出處，本文鏈接：https://www.uj5u.com/caozuo/455956.html