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,,我对如何自己实现此想法有所了解:
from 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解决了这一问题.
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。