PySpark读取HDFS文件执行collect时遇Java堆内存溢出问题求助

Hey there, let's work through this OutOfMemoryError you're hitting when calling collect() on your PySpark RDD. I'll break down the root causes and walk you through actionable fixes.

First, Spot the Quick Code Mistake #

Looking at your code, you have fileRDD.cache — notice the missing parentheses? This means the cache action never actually runs. You need to update it to:

fileRDD = sc.textFile("/tmp/test_file.txt").cache()

(Reassigning the RDD ensures you're using the cached version for subsequent operations.)

Why collect() Is Causing OOM #

The core issue here is that collect() pulls all data from every Executor node directly into the Driver node's memory. Even though your Executors have plenty of memory (72g each), your Driver only has 16g allocated, and spark.driver.maxResultSize is set to 10000 (which I assume is ~10GB). When you convert the 500MB text file into a list of Python lists, the data expands significantly in memory — easily exceeding that 10GB limit.

Fixes to Try (Ordered by Best Practice) #

1. Avoid collect() Entirely (Optimal Solution)

Instead of pulling all data to the Driver to build your numpy matrix, process data in parallel on Executors first, then only pull small intermediate results to the Driver. For example, build smaller numpy matrices per partition and merge them on the Driver:

import numpy as np

def partition_to_matrix(iterator):
    # Convert a partition's lines into a numpy array
    lines = list(iterator)
    matrix_chunk = np.array([list(map(float, line.split())) for line in lines])
    return [matrix_chunk]

# Generate matrix chunks on each Executor
partition_chunks = fileRDD.mapPartitions(partition_to_matrix).collect()

# Merge chunks into the final matrix on the Driver
final_matrix = np.concatenate(partition_chunks, axis=0)

This way, you're only collecting compact numpy arrays instead of bloated Python lists, drastically reducing Driver memory usage.

2. Tune Driver Memory Limits

If you absolutely need to use collect() with the full list of lines, adjust these two configs to give the Driver more breathing room:

conf = SparkConf().setAppName('test') \
 .set('spark.executor.cores', 4) \
 .set('spark.executor.memory', '72g') \
 .set('spark.driver.memory', '24g')  # Increase Driver memory (match your cluster's available resources)
 .set('spark.yarn.executor.memoryOverhead', 4096) \
 .set('spark.dynamicAllocation.enabled', 'true') \
 .set('spark.shuffle.service.enabled', 'true') \
 .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
 .set('spark.driver.maxResultSize', '18g')  # Set to ~75% of driver memory to avoid OOM
 .set('spark.kryoserializer.buffer.max', 2044)

Just make sure your cluster's Driver node has enough physical memory to support this increase.

3. Optimize RDD Partitioning

While not the main issue here, adjusting the number of partitions can help with memory pressure during processing. You can explicitly set partitions when reading the file:

fileRDD = sc.textFile("/tmp/test_file.txt", minPartitions=16).cache()

Smaller partitions make data processing more granular, though this won't fix the Driver OOM on its own.

4. Switch to DataFrames for Better Memory Efficiency

PySpark DataFrames use more efficient serialization (especially with Kryo) and memory management than raw RDDs. Converting your data to a DataFrame can reduce memory footprint:

from pyspark.sql import SparkSession
from pyspark.sql.functions import split

spark = SparkSession.builder.config(conf=conf).getOrCreate()
df = spark.read.text("/tmp/test_file.txt")
# Split lines and convert to numeric types as needed
df = df.withColumn("split_line", split(df.value, " "))
# Process further without collecting, or collect smaller subsets if needed

Final Notes #

Always prioritize processing data on Executors instead of pulling everything to the Driver — that's the whole point of using Spark! The collect() method should only be used for small datasets or final results that fit comfortably in Driver memory.

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