PySpark行访问与转换优化：S3大数据ETL串行执行问题及优化

咱们先直接明确：这段代码完全是串行执行的。原因在于df.rdd.toLocalIterator()会把RDD的每个分区依次拉到Driver节点本地，然后你用单线程的for循环逐行处理——等于把Spark集群的分布式能力完全浪费了，所有计算都挤在Driver上，5GB的数据这么跑不仅慢，还很容易因为Driver内存不足导致崩溃。

为什么你的写法效率低？ #

• 把分布式存储的集群数据拉到单节点Driver处理，完全没利用Spark的并行计算能力
• 逐行写入S3，会产生上万个甚至更多小文件，S3对大量小文件的读写性能差，还会增加请求次数成本
• 手动遍历每行删除空值列，这种循环操作在Python里本身就慢，还占用Driver资源

优化方案：利用Spark分布式能力，避免本地串行处理 #

1. 用Spark分布式API处理空值，替代Driver端逐行遍历 #

不需要把数据拉到本地，直接在集群的Executor上并行处理每行数据。这里分两种场景：

场景1：删除每行中值为null的字段（和你代码的逻辑一致）

可以用RDD的map操作（分布式执行），或者Spark SQL的UDF来实现：

import json
from pyspark.sql import functions as F

# 方式一：用RDD API分布式处理
processed_rdd = df.rdd.map(lambda row: json.dumps({
    k: v for k, v in row.asDict(True).items() if v is not None
}))
# 批量写入S3，Spark会自动按分区生成文件
processed_rdd.saveAsTextFile("s3://write-test-transaction-transformed/")

# 方式二：用Spark SQL UDF处理
def clean_row(row):
    return json.dumps({k: v for k, v in row.asDict(True).items() if v is not None})

clean_udf = F.udf(clean_row, F.StringType())
df.withColumn("cleaned_json", clean_udf(F.struct(df.columns))) \
  .select("cleaned_json") \
  .write.mode("overwrite") \
  .text("s3://write-test-transaction-transformed/")

场景2：删除整列全为空的字段

如果你的需求是删除整个列都为null的字段，直接用Spark内置的dropna更高效：

# 删除所有值都是null的列
cleaned_df = df.dropna(how="all", axis=1)
# 写入S3
cleaned_df.write.mode("overwrite").json("s3://write-test-transaction-transformed/")

2. 控制S3写入的文件数量，避免大量小文件 #

Spark默认会按分区数生成文件，你可以通过repartition或coalesce调整分区数，比如5GB的数据可以设置成40个分区（大概128MB/分区，符合S3的最优文件大小）：

# 调整分区数后写入
processed_rdd.repartition(40).saveAsTextFile("s3://write-test-transaction-transformed/")
# 或者DataFrame写法
cleaned_df.repartition(40).write.mode("overwrite").json("s3://write-test-transaction-transformed/")

更优的实现思路 #

1. 彻底抛弃Driver端串行处理：Spark的核心是分布式计算，所有数据处理逻辑都应该放在Executor上执行，尽量减少Driver和Executor之间的数据传输。
2. 优先用Spark内置API：内置API（比如dropna、to_json等）是经过优化的，比手动写Python循环效率高得多。
3. 尽量用列式存储格式：如果后续还需要用Spark或其他工具处理数据，建议把数据写成Parquet格式——它是列式存储，比JSON更节省空间，查询效率也更高：

# 处理空值后写入Parquet
cleaned_df.write.mode("overwrite").parquet("s3://write-test-transaction-transformed/")

4. 避免逐行写入S3：S3适合大文件读写，批量写入不仅性能更好，还能降低请求成本。

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