PySpark 数据框 Pandas UDF 返回空数据框答案

【问题标题】：PySpark dataframe Pandas UDF returns empty dataframePySpark 数据框 Pandas UDF 返回空数据框
【发布时间】：2020-09-03 12:54:23
【问题描述】：

我正在尝试按照groupby('Key').apply(UDF) 方法将pandas_udf 应用于我的PySpark 数据帧以进行一些过滤。要使用pandas_udf，我定义了一个输出schema，并在Number 列上有一个条件。例如，这里的简化想法是我只希望返回具有奇数Number 的行的ID。

这带来了一个问题，有时组中没有奇怪的Number，因此 UDF 只返回一个空数据帧，这与定义的 schema 冲突以返回 int 为 Number .

有没有办法解决这个问题，只输出和组合所有奇数 Number 行作为一个新的数据帧？

schema = StructType([
        StructField("Key", StringType()),
        StructField("Number", IntegerType())
    ])

@pandas_udf(schema, functionType=PandasUDFType.GROUPED_MAP)
    def get_odd(df):
        odd = df.loc[df['Number']%2 == 1]
        return odd[['ID', 'Number']]

【问题讨论】：

因为我希望将算法部署在集群上，而 groupby 可以实现分布式计算。在没有 groupby 的情况下，在一个巨大的数据帧上执行我的条件非常昂贵。
使用 if/else 返回一个定义了列的空数据框？另外，您的返回如何与架构匹配，因为您只返回了 ID 列？
打错字了，刚刚改正。

标签： pandas dataframe apache-spark pyspark

【解决方案1】：

我在某些组中遇到了 null DataFrame 的问题。我通过检查空 DataFrame 并返回一个定义了模式的 DataFrame 来解决这个问题：

if df_out.empty:
    # change the schema as needed
    return pd.DataFrame({'fullVisitorId': pd.Series([], dtype='str'),
                         'time': pd.Series([], dtype='datetime64[ns]'),
                         'total_transactions': pd.Series([], dtype='int')})

【讨论】：

我发现单独提供列名就足够了。因此，如果我的 pandas_udf 函数出现问题并且我想返回一个空的 pandas 数据框，我只需这样做：return pd.DataFrame(columns=schema.fieldNames())，其中schema 是您传递给 @ 的 Spark DataFrame 的架构（将被返回） 987654325@函数。