This answer很好地解释了如何使用pyspark的groupby和pandas_udf进行自定义聚合.但是,我不可能如示例的这一部分所示手动声明我的模式
from pyspark.sql.types import *
schema = StructType([
StructField("key",StringType()),StructField("avg_min",DoubleType())
])
因为我将返回100列具有自动生成名称的列.是否有任何方法告诉PySpark隐式使用我的函数返回的Schema,并假定所有工作节点的Schema都相同？由于我将不得不使用我要使用的预测变量,因此该模式在运行期间也将发生变化,因此可以自动选择生成模式的流程…
最佳答案
基于Sanxofons comment,,我对如何自己实现此想法有所了解：
原文链接：https://www.f2er.com/python/533145.htmlfrom pyspark.sql.types import *
mapping = {"float64": DoubleType,"object":StringType,"int64":IntegerType} # Incomplete - extend with your types.
def createUDFSchemaFromPandas(dfp):
column_types = [StructField(key,mapping[str(dfp.dtypes[key])]()) for key in dfp.columns]
schema = StructType(column_types)
return schema
我要做的是获取一个示例熊猫df,将其传递给函数,然后查看返回的结果：
dfp = df_total.limit(100).toPandas()
df_return = my_UDF_function(dfp)
schema = createUDFSchemaFromPandas(df_return)
这似乎为我工作.问题是它是递归的(需要定义函数以获取模式,将模式定义为udf).我通过创建一个简单地传递数据帧的“包装” UDF解决了这一问题.