带有groupby的pyspark collect_set或collect_list

Question

在groupby之后，如何在数据帧上使用collect_set或collect_list。例如：df.groupby('key').collect_set('values')。我得到一个错误：AttributeError: 'GroupedData' object has no attribute 'collect_set'

Answer 1

您需要使用agg。示例：

from pyspark import SparkContext
from pyspark.sql import HiveContext
from pyspark.sql import functions as F

sc = SparkContext("local")

sqlContext = HiveContext(sc)

df = sqlContext.createDataFrame([
    ("a", None, None),
    ("a", "code1", None),
    ("a", "code2", "name2"),
], ["id", "code", "name"])

df.show()

+---+-----+-----+
| id| code| name|
+---+-----+-----+
|  a| null| null|
|  a|code1| null|
|  a|code2|name2|
+---+-----+-----+

注意，在上面的代码中，您必须创建一个HiveContext。请参阅https://stackoverflow.com/a/35529093/690430，了解如何处理不同的Spark版本。

(df
  .groupby("id")
  .agg(F.collect_set("code"),
       F.collect_list("name"))
  .show())

+---+-----------------+------------------+
| id|collect_set(code)|collect_list(name)|
+---+-----------------+------------------+
|  a|   [code1, code2]|           [name2]|
+---+-----------------+------------------+

Answer 2

如果你的数据帧很大，你可以尝试使用pandas udf(GROUPED_AGG)来避免内存错误。它的速度也快得多。

分组聚合Pandas UDF类似于Spark聚合函数。分组聚合Pandas UDF与groupBy()、.agg()和pyspark.sql.Window一起使用。它定义了从一个或多个pandas.Series到标量值的聚合，其中每个pandas.Series表示组或窗口中的一列。pandas udf

示例：

import pyspark.sql.functions as F

@F.pandas_udf('string', F.PandasUDFType.GROUPED_AGG)
def collect_list(name):
    return ', '.join(name)

grouped_df = df.groupby('id').agg(collect_list(df["name"]).alias('names'))