我想運行一個帶有初始化代碼的 pyspark UDF，每個 Python 行程只運行一次。我不需要保證這段代碼只運行一次，但為了性能，我不希望它為每一行運行。
在斯卡拉我可以這樣做：
object MyUDF {
// This code runs once per JVM (executor), initialize heavy objects
}
class MyUDF extends UDF2[Double, Double, String] {
override def call(lat: Double, long: Double): String = {
// This code runs per record,
but can use the static objects that are already initialized
}
}
Python有類似的可能嗎？我知道每個執行程式都有一個運行的 Python 行程來回應該執行程式的 UDF 呼叫。
uj5u.com熱心網友回復：
讓我們考慮一個玩具示例，其中大物件是一個串列，您的 UDF 只是檢查一個元素是否在該串列中。
一種方法是簡單地在 UDF 定義之外定義物件。因此，初始化代碼將只在驅動程式中執行一次。
from pyspark.sql import functions as f
big_list = [1, 2, 4, 6, 8, 10]
is_in_list = f.udf(lambda x: x in big_list)
spark.range(10).withColumn("x", is_in_list(f.col("id"))).show()
產生：
--- -----
| id| x|
--- -----
| 0|false|
| 1| true|
| 2| true|
| 3|false|
| 4| true|
| 5|false|
| 6| true|
| 7|false|
| 8| true|
| 9|false|
--- -----
該代碼以及您的 scala 代碼的問題在于， spark 將在每個任務中發送該物件的副本。如果您關心的只是初始化物件所需的時間，那很好。但是如果物件很大，可能會影響作業的性能。
要解決這個問題，您可以使用廣播變數。事實上，根據 spark 的檔案：
廣播變數允許程式員在每臺機器上快取一個只讀變數，而不是隨任務一起發送它的副本。
代碼將非常相似：
from pyspark.sql import functions as f
big_list = [1, 2, 4, 6, 8, 10]
big_list_bc = sc.broadcast(big_list)
is_in_list = f.udf(lambda x: x in big_list_bc.value)
spark.range(10).withColumn("x", is_in_list(f.col("id"))).show()
uj5u.com熱心網友回復：
幾種型別之一pandas_udf是 Iterator[pd.Series]) -> Iterator[pd.Series]。
在映射每個 pd.Series 并產生結果之前，您可以有一個 init 階段，該階段將在每個執行程式中運行一次。
可以在此處找到有關該主題的相當好的指南。
@pandas_udf("long")
def plus_one(batch_iter: Iterator[pd.Series]) -> Iterator[pd.Series]:
<init something>
for x in batch_iter:
yield <your code here>
轉載請註明出處，本文鏈接：https://www.uj5u.com/qiye/486887.html
上一篇：使用數學在網格內查找專案的位置