我已經搜索并找不到適合我的 Pyspark 問題的答案。我正在尋找一種更有效且不使用 UDF 的替代方法。
我在 UDF 中有一個簡單的方程，它的輸入來自 (a) 文字常量、(b) 列值和 (c) 來自串列（或 dict）的值。輸出必須多次創建并存盤在陣列中。是否可以在 UDF 之外執行此操作？
我敲了這個簡單的例子，雖然我的實際問題稍微復雜一些，行數更多，方程更大，& 回圈超過 40 次：
注意：V3 示例問題：
from pyspark.sql.functions import *
from pyspark.sql.types import *
test_data = [("A1",10.5), ("A2",40.5), ("A3",60.5)]
schema = StructType([ \
StructField("ID",StringType(),True), \
StructField("num1",DoubleType(),True)])
df = spark.createDataFrame(data=test_data,schema=schema)
const1 = 10
const2 = 20
num_lst1 = [2.1,4.2,6.3,8.4,10.5]
num_lst2 = [20,40,60,80,100]
num_lst3 = [100.1,200.2,300.3,400.4,500.5]
def udf_whatever(num_lst1,num_lst2,num_lst3):
def whatever(const1, const2, val1):
DH = [None for t in range(5)]
for i in range(5):
DH[i] = const1 val1 const2 (num_lst1[i]*num_lst2[i]) num_lst3[i]
return DH
return udf(whatever, ArrayType(DoubleType()))
df2 = df.withColumn("UDF_OUT",udf_whatever(num_lst1,num_lst2,num_lst3)(lit(const1),lit(const2),col("num1")))
df2.show(truncate=False)
--- ---- -------------------------------------
|ID |num1|UDF_OUT |
--- ---- -------------------------------------
|A1 |10.5|[182.6, 408.7, 718.8, 1112.9, 1591.0]|
|A2 |40.5|[212.6, 438.7, 748.8, 1142.9, 1621.0]|
|A3 |60.5|[232.6, 458.7, 768.8, 1162.9, 1641.0]|
--- ---- -------------------------------------
在 Emma 的幫助下（在評論中），我已經完成了這項作業，但為每個串列創建新列似乎有點貴，尤其是在數百萬行的情況下。有沒有更好的辦法？
df3 = df.withColumn('MAP_LIST1', array(*map(lit, num_lst1)))\
.withColumn('MAP_LIST2', array(*map(lit, num_lst2)))\
.withColumn('MAP_LIST3', array(*map(lit, num_lst3)))\
.withColumn('EQUATION_OUT', expr(f"""transform(MAP_LIST1, (x, i) -> {const1} num1 {const2} (x * MAP_LIST2[i]) MAP_LIST3[i])"""))
df3.show()
非常感謝任何幫助！瑞克
uj5u.com熱心網友回復：
一種方法是使用array_repeat和transform。
首先，用于array_repeat創建僅包含num3值的基本陣列。
然后，用于計算陣列transform中每個值的值。num3
對于 Spark 3.1
repeat = 5
const = 10
df = (df.withColumn('arr', array_repeat('num3', repeat))
.withColumn('arr', transform(col('arr'), lambda x, i: lit(const) col('num1') col('num2') i * x)))
對于 Spark 2.4 < 3.1
df = (df.withColumn('arr', array_repeat('num3', repeat))
.withColumn('arr', expr('transform(arr, (x, i) -> 10 num1 num2 i * x)')))
==================================================== ==========
用新方程更新（const col list element）
如果只有 1 個陣列（ ），您可以用陣列num_lst初始化并添加其他變數到.UDF_OUTtransformUDF_OUT
df = (df.withColumn('UDF_OUT', array(*map(lit, num_lst)))
.withColumn('UDF_OUT', expr(f"""
transform(UDF_OUT, (x, i) -> {const} num1 x)
""")))
轉載請註明出處，本文鏈接：https://www.uj5u.com/yidong/484057.html