在 spark 数据帧上实现 pythonic 统计函数

Question

我在跨节点分布的 spark 数据帧中有非常大的数据集。 我可以使用 spark 库 pyspark.sql.functions 进行简单的统计，例如 mean、stdev、skewness、kurtosis 等。

如果我想使用 Jarque-Bera (JB) 或 Shapiro-Wilk(SW) 等高级统计测试，我会使用 scipy 等 python 库，因为标准 apache pyspark 库没有它们。但为了做到这一点，我必须将 spark 数据帧转换为 pandas，这意味着像这样强制数据进入主节点：

import scipy.stats as stats
pandas_df=spark_df.toPandas()
JBtest=stats.jarque_bera(pandas_df)
SWtest=stats.shapiro(pandas_df)

我有多个特征，每个特征 ID 对应一个我要对其执行测试统计的数据集。

我的问题是：

有没有办法在数据仍然分布在节点上的同时将这些 pythonic 函数应用于 spark 数据帧，还是我需要在 spark 中创建自己的 JB/SW 测试统计函数？

感谢您提供宝贵的见解

Answer 1

您应该能够定义一个包装 Pandas 函数 (https://databricks.com/blog/2017/10/30/introducing-vectorized-udfs-for-pyspark.html) 的矢量化用户定义函数，如下所示：

from pyspark.sql.functions import pandas_udf, PandasUDFType
import scipy.stats as stats

@pandas_udf('double', PandasUDFType.SCALAR)

def vector_jarque_bera(x):
    return stats.jarque_bera(x)

# test:
spark_df.withColumn('y', vector_jarque_bera(df['x']))

请注意，向量化函数 column 以一列作为其参数并返回一列。

（注意。@pandas_udf 装饰器将其下方定义的 Pandas 函数转换为向量化函数。返回向量的每个元素本身都是一个标量，这就是传递参数 PandasUDFType.SCALAR 的原因。）