如何在PySpark中定义用户定义的聚合函数

Question

我想在pyspark中创建一个用户定义的聚合函数。我找到了一些Scala的文档，并希望在Python中实现类似的功能。

更具体地说，假设我已经实现了这样的功能：

def process_data(df: pyspark.sql.DataFrame) -> bytes:
...  # do something very complicated here

现在我希望能够做一些类似的事情：

source_df.groupBy("Foo_ID").agg(UDAF(process_data))

现在的问题是，我应该用什么来代替UDAF？

Answer 1

PySpark不直接支持UDAF，所以我们必须手动进行聚合。

参考：

1. https://spark.apache.org/docs/latest/sql-ref-functions-udf-aggregate.html

2. 如何在多列上编写Pyspark UDAF？

3. 在PySpark中的GroupedData上应用UDF(以python为例(

4. https://florianwilhelm.info/2017/10/efficient_udfs_with_pyspark/