我正在嘗試創建一個將另一個函數作為參數的 UDF。但是執行以異常結束。我運行的代碼：import pandas as pdfrom pyspark import SparkConf, SparkContext, SQLContextfrom pyspark.sql.types import MapType, DataType, StringTypefrom pyspark.sql.functions import udf, struct, litimport ossc = SparkContext.getOrCreate(conf=conf)sqlContext = SQLContext(sc)df_to_test = sqlContext.createDataFrame( pd.DataFrame({ 'inn': ['111', '222', '333'], 'field1': [1, 2, 3], 'field2': ['a', 'b', 'c'] }))def foo_fun(row, b) -> str: return 'a' + b()def bar_fun(): return 'I am bar'foo_fun_udf = udf(foo_fun, StringType())df_to_test.withColumn( 'foo', foo_fun_udf(struct([df_to_test[x] for x in df_to_test.columns]), bar_fun)).show()例外：Invalid argument, not a string or column: <function bar_fun at 0x7f0e69ce6268> of type <class 'function'>. For column literals, use 'lit', 'array', 'struct' or 'create_map' function.我試圖包裝bar_fun成 udf 但沒有成功。有沒有辦法將函數作為參數傳遞？
1 回答
墨色風雨
TA貢獻1853條經驗 獲得超6個贊
你離解決方案不遠了。這是我會怎么做：
def foo_fun_udf(func):
def foo_fun(row) -> str:
return 'a' + func()
out_udf = udf(foo_fun, StringType())
return out_udf
df_to_test.withColumn(
'foo',
foo_fun_udf(bar_fun)(struct([df_to_test[x] for x in df_to_test.columns]))
).show()
添加回答
舉報
0/150
提交
取消