python-PySpark中pandas_udf的隐式架构？

发布时间：2020-12-17 17:37:52 所属栏目：Python 来源：网络整理

导读：This answer很好地解释了如何使用pyspark的groupby和pandas_udf进行自定义聚合.但是,我不可能如示例的这一部分所示手动声明我的模式 from pyspark.sql.types import *schema = StructType([ StructField("key",StringType()),StructField("avg_min",DoubleTy

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解决了这一问题.

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!