使用PySpark将大CSV按每100行分组发送至Lambda的方案咨询

原有方案的核心问题 #

你当前的实现调用了collect()，会把1亿+行全量数据全部拉取到Spark驱动节点内存，生产环境运行100%会出现OOM溢出，完全无法处理超大规模CSV文件。

最优实现方案 #

推荐直接使用foreachPartition按分区处理，数据全程在executor节点分布式加载，不会落到驱动节点，完美适配超大数据量场景，逻辑也更简洁。

方案1：RDD实现（和你现有技术栈对齐） #

def send_batch_to_lambda(iterator):
    buffer = []
    for row in iterator:
        buffer.append(row)
        # 凑够100行就发送
        if len(buffer) == 100:
            # 此处替换为你实际调用Lambda的逻辑
            print("Send batch:", buffer)
            buffer = []
    # 处理分区末尾不足100行的剩余数据
    if buffer:
        print("Send remaining batch:", buffer)

# 全流程无驱动节点数据拉取，分布式执行
sc.textFile(csv_file).foreachPartition(send_batch_to_lambda)

方案2：DataFrame实现（如果需要结构化处理CSV） #

如果你需要提前对CSV做字段解析、过滤、转换等预处理，用SparkSession的DataFrame API更方便：

from pyspark.sql import SparkSession
import itertools

spark = SparkSession.builder.appName("csv_batch_process").getOrCreate()

# 可添加header=True、指定schema参数直接解析结构化CSV
df = spark.read.csv(csv_file)

def process_df_partition(iterator):
    # 直接用itertools工具按100行切分批次
    for batch in itertools.batched(iterator, 100):
        # 替换为Lambda调用逻辑
        print(f"Send batch size: {len(batch)}")

df.foreachPartition(process_df_partition)

额外优化建议 #

• 可以通过repartition(并发数)调整分区数，匹配你Lambda的并发上限，提升整体处理速度，示例：sc.textFile(csv_file).repartition(100).foreachPartition(...)
• 调用Lambda时建议使用异步调用，避免单批阻塞影响整体作业效率
• 只有当你需要批次全局有序时，才需要额外加索引排序处理，否则上面的无序分批方案性能最优

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