python - 优化Pyspark的Collect_List函数

Question

我需要聚合我的数据，以便生成以下输出:

JSON 输出

{
    "people": [
       {
         "firstName": "Jimi",
         "lastName": "Hendrix",
         "age": "27"
       },
       {
         "firstName": "Jimmy",
         "lastName": "Page",
         "age": "75"
       }
    ]
}

但是，当运行聚合函数(如下)时，我收到此错误:

Caused by: org:apache.spark.SparkSession: Job aborted due to stage failure: Total size of serialized results of <task_size> task (20GB) is bigger than spark.driver.maxResultSize (20.0 GB)

This leads me to believe that the collect_list function is what's causing this problem. Instead of running the tasks in parallel, they run on a single node and run out of memory.

创建 JSON 输出的最佳方法是什么？有没有办法优化 collect_list 函数？

示例代码:

def aggregate(df):
    return df.agg(collect_list(struct(
        df.firstName,
        df.lastName,
        df.age
    )).alias('people'))

Answer 1

当您执行收集列表时，所有数据都会作为列表收集到驱动程序。相反，您可以使用 JSON 在数据框中创建一列并将其存储为 CSV。

替代方法是增加驱动程序内存 -

conf.set("spark.driver.maxResultSize", "25g")