如何在AWS Glue 4.0中并行写入PySpark DataFrame列表至工作节点？

问题本质 #

你遇到的两个问题核心原因：

• 遍历列表串行写入：驱动端按顺序执行每个DF的写入操作，每个作业必须等前一个完成才会启动，完全没有利用集群并行能力。
• SparkContext.parallelize/forEach报错：Spark的worker节点无法访问驱动端的SparkContext或SparkSession，而DataFrame依赖驱动端的会话上下文，在worker中操作DF必然触发RuntimeError。

可行解决方案 #

方案1：驱动端多线程提交并行写入作业 #

在驱动端启动多个线程，每个线程独立提交一个DataFrame的写入任务，Spark集群会并行处理这些独立的写入作业，这是最直接有效的方法。

示例代码：

import threading
from pyspark.sql import DataFrame

def write_df(df: DataFrame, output_path: str):
    # 可根据需求调整写入格式和模式
    df.write.mode("overwrite").parquet(output_path)

# 假设你有DataFrame集合和对应的输出路径列表
df_list = [df1, df2, df3, ...]
output_paths = ["s3://bucket/path1", "s3://bucket/path2", "s3://bucket/path3", ...]

# 启动线程并行执行写入
threads = []
for df, path in zip(df_list, output_paths):
    thread = threading.Thread(target=write_df, args=(df, path))
    thread.start()
    threads.append(thread)

# 等待所有写入任务完成
for thread in threads:
    thread.join()

注意事项：

• 控制并发线程数：根据Glue集群的Worker数量和资源配置调整并发数，避免同时提交过多作业导致资源耗尽或队列阻塞。
• 写入模式：根据业务需求选择overwrite/append/ignore等模式，避免数据冲突。

方案2：合并同结构DataFrame后分区写入（仅适用于结构相同的DF） #

如果所有DataFrame的Schema完全一致，可以将它们合并为一个大的DataFrame，添加一个分区字段后按分区写入，利用Spark原生的分布式写入能力实现并行效果。

示例代码：

from pyspark.sql import functions as F

# 给每个DF添加标识字段，用于后续分区
df1 = df1.withColumn("df_id", F.lit("df1"))
df2 = df2.withColumn("df_id", F.lit("df2"))
df3 = df3.withColumn("df_id", F.lit("df3"))

# 合并所有DF
combined_df = df1.unionByName(df2).unionByName(df3)

# 按标识字段分区写入，Spark会并行处理不同分区的写入
combined_df.write.mode("overwrite").partitionBy("df_id").parquet("s3://bucket/combined_path")

为什么之前的方法不可行 #

• 串行遍历：驱动端单线程依次执行df.write，每个写入作业必须等待前一个完成，完全没有并行性。
• parallelize/forEach：parallelize将本地DF列表转为RDD后，forEach的逻辑在worker节点执行，但worker无法访问驱动端的SparkSession，DataFrame无法在worker环境中被序列化和操作，因此触发RuntimeError。

内容的提问来源于stack exchange，提问作者daniel9x