Why does collect() crash my driver?

Summary

collect() returns every element of an RDD or DataFrame to the driver as one array or list, so the whole result has to fit on a single machine. The Spark docs warn that this can make the driver run out of memory, and spark.driver.maxResultSize (default 1g) aborts the job when the serialized results exceed it. Fix it by returning less data, or by writing large results to storage instead of collecting them.

Why does collect() crash my driver?

collect() returns all the elements of the dataset to the driver program as an array, or as a list in PySpark. The RDD programming guide warns that this "can cause the driver to run out of memory, though, because collect() fetches the entire RDD to a single machine." Adding executors does not help, because the result still lands on one machine, the driver. The PySpark docs repeat the warning for RDD.collect(), RDD.take(), DataFrame.collect() and DataFrame.toPandas(): use them only if the result is expected to be small, "as all the data is loaded into the driver's memory."

What fails, and how do I fix it?

Symptom	Likely cause	Fix
Job aborted due to stage failure: Total size of serialized results of <N> tasks (<size>) is bigger than spark.driver.maxResultSize (<limit>)	The serialized results of all partitions exceeded spark.driver.maxResultSize, which defaults to 1g	Collect less data. Raise the limit only if the workload requires it
The driver runs out of memory	A large result, plus the memory overhead of objects in the JVM, does not fit in spark.driver.memory, which defaults to 1g	Collect less data, or use a larger driver
A BI tool using JDBC or ODBC fails with the same error on Databricks	The result set is routed through driver memory	Enable Cloud Fetch

How do I collect less data?

Option	When to use it	Trade-off
take(n) or df.show(n)	Look at a few rows	take also loads its result into driver memory, so keep n small
df.limit(n)	Keep working with a DataFrame	It returns a DataFrame and collects nothing
Filter or aggregate, then collect()	The reduced result is small	You have to be sure it really is small
df.write.parquet(path) or rdd.saveAsTextFile(path)	The result is large	Output goes to storage, not to the driver
df.toLocalIterator()	You need every row on the driver	Uses as much memory as the largest partition

df.show()                               # n defaults to 20
first_rows = df.take(5)                 # a list of Row, keep n small
preview = df.limit(100)                 # a DataFrame, nothing collected yet

counts = df.groupBy("region").count().collect()   # aggregate first

df.write.parquet("/data/output")        # large results go to storage
rdd.saveAsTextFile("/data/output_txt")  # the RDD equivalent

for row in df.toLocalIterator():        # memory of the largest partition
    handle(row)

Should I raise spark.driver.maxResultSize or spark.driver.memory?

Treat both as last resorts. The configuration docs say: "Having a high limit may cause out-of-memory errors in driver ... Setting a proper limit can protect the driver from out-of-memory errors." A value of 0 means unlimited. If you do need a bigger driver, note that spark.driver.memory defaults to 1g, and in client mode it "must not be set through the SparkConf directly in your application, because the driver JVM has already started at that point." Set it with the --driver-memory command line option or in your default properties file.

How Databricks approaches this

The Databricks knowledge base article on the maxResultSize exception gives the same order of fixes. First "refactor your code to prevent the driver node from collecting a large amount of data." Only if the workload requires it, raise spark.driver.maxResultSize in the compute's Spark config, and note the warning that very high values "can lead to OOM errors." The compute docs also let you choose a larger driver node type "if you are planning to collect() a lot of data from Spark workers and analyze them in the notebook," and advise detaching unused notebooks from the driver. To see whether the driver is the node under pressure, open the compute's Metrics tab. By default it shows metrics for all nodes, including the driver, averaged over the time period, and the All nodes drop-down lets you select a single node. For BI tools over JDBC or ODBC, Cloud Fetch has executors upload result sets to cloud storage and the driver hand out pre-signed URLs, "avoiding the driver having to use up its memory." The Spark Connect docs say that "Applications that use too much memory will now only impact their own environment as they can run in their own processes."

FAQs

What is spark.driver.maxResultSize?

It is the limit of the total size of serialized results of all partitions for each Spark action (for example collect), in bytes. The default is 1g, it should be at least 1M, and 0 means unlimited. Jobs are aborted if the total size is above the limit.

Does this happen with DataFrames too?

Yes. The DataFrame.collect() docs carry the same warning: use it only if the resulting list is expected to be small. DataFrame.toPandas() carries it too. For samples, use limit(), take() or show().

Is take(n) always safe?

Not when n is large. RDD.take() works by first scanning one partition and using that result to estimate how many more partitions it needs, and it also loads its result into driver memory.