PySpark处理684GB大文件count()操作超时终止问题求助

问题背景 #

本地环境16GB内存、4核CPU，使用PySpark 3.3.0处理684GB TSV文件，执行groupBy('timestamp_day').count()时耗时过久最终进程终止，当前数据分区数为5480。

核心原因 #

1. 本地资源有限，默认Spark配置未适配大文件处理，内存分配不足
2. 初始分区数过多（5480），导致任务调度开销巨大，shuffle阶段数据传输压力陡增
3. 直接groupBy触发全量shuffle，未做局部聚合优化，内存负载远超硬件承载能力

解决方案 #

1. 调整SparkSession资源与核心配置 #

创建Session时明确分配内存、适配硬件调整核心参数，避免资源耗尽：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName('pyspark-shellTest2') \
    .master('local[3]')  # 留1核给系统，避免CPU满载
    .config('spark.driver.memory', '10g')  # 分配10G内存给Driver，剩余6G给系统进程
    .config('spark.driver.memoryOverhead', '2g')  # 预留内存开销，防止OOM
    .config('spark.sql.shuffle.partitions', '100')  # 减少shuffle分区数，降低调度成本
    .config('spark.sql.files.maxPartitionBytes', '256m')  # 增大单分区文件大小，减少初始分区数
    .config('spark.executor.extraJavaOptions', '-XX:+UseG1GC')  # 用G1垃圾回收，减少GC停顿
    .getOrCreate()

2. 读取数据时仅加载必要列 #

统计仅需timestamp_day列，丢弃user_id和course_id可大幅降低内存占用：

# 只读取目标列，减少内存消耗
raw_data = spark.read.options(delimiter="\t", header=True).csv("O:/Corbin/Canvas/requests_12_05_2022.txt")
timestamp_data = raw_data.select('timestamp_day')

3. 优化聚合逻辑：局部聚合替代全量shuffle #

用RDD的reduceByKey先在分区内完成局部计数，再全局合并，大幅减少shuffle数据量：

# 先分区内聚合，再全局合并，shuffle数据量仅为groupBy的1/N（N为分区数）
count_result = timestamp_data.rdd \
    .map(lambda row: (row.timestamp_day, 1)) \
    .reduceByKey(lambda a, b: a + b) \
    .toDF(['timestamp_day', 'count'])

count_result.show()

4. 合并过多分区（可选） #

如果初始分区数仍过多，用coalesce无shuffle合并分区，降低调度开销：

# 合并分区到合理数量（比如200，根据CPU核数调整）
timestamp_data = timestamp_data.coalesce(200)
# 再执行聚合操作
timestamp_data.groupBy('timestamp_day').count().show()

关键说明 #

• 本地模式下Driver与Executor共享资源，重点优化Driver内存配置
• reduceByKey比DataFrame原生groupBy.count()更适合大文件计数，避免全量数据shuffle
• 分区大小建议控制在128MB-512MB之间，平衡调度开销与内存负载

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