Spark Configurations

A setting is a named value that tells Spark how to run, such as spark.executor.memory. Spark has hundreds of settings.

Only a few decide if a job is fast or slow. This chapter shows those few, and how to work out their values from your data and your cluster.

You already know the parts. An executor is a process on a worker machine that runs tasks. A partition is one piece of the data.

You know the memory model from "Memory Management & Tungsten". You know how to read the Spark UI from "Performance Tuning & Debugging with the Spark UI". Now you set the numbers.

The chapter has seven parts.

1. Where you can set a value, and which place wins.
2. Which settings cannot change after the job starts.
3. The formula for executors, cores and memory.
4. A full example on a real-size cluster.
5. How to choose the number of shuffle partitions.
6. How the numbers change with the size of your data.
7. Dynamic allocation: adding and removing executors.

Every number in this chapter is a starting point. Run the job, read the Spark UI, then adjust.

You can set the same setting in three places. If they disagree, Spark picks one by a fixed order.

Here is the order, from the strongest place to the weakest.

1. In your code: SparkSession.builder.config(...) before the session starts, or spark.conf.set(...) for SQL settings. This wins.
2. In the spark-submit command: --conf key=value, or an option such as --executor-memory.
3. In the file spark-defaults.conf. It holds defaults for every job on the cluster.

If no place sets a value, Spark uses its built-in default. Look at one setting that is set in all three places.

# In spark-defaults.conf:
#   spark.sql.shuffle.partitions   100
# In the spark-submit command:
#   --conf spark.sql.shuffle.partitions=400
# In the code:
spark = (SparkSession.builder
         .config("spark.sql.shuffle.partitions", "8")
         .getOrCreate())

print(spark.conf.get("spark.sql.shuffle.partitions"))

To see what is set at launch, open the Environment tab of the Spark UI. It lists only values set by you or by Spark, and a missing setting uses its default. For values changed later with spark.conf.set, use spark.conf.get.

Some settings are fixed when the application starts. Resource settings are the main case: executor memory, executor cores and the number of executors.

The reason is simple. Spark fixes the size of every executor when the application starts. A cluster manager, such as YARN or Kubernetes, hands out the machines.

SQL settings are different. You can change them at any time with spark.conf.set. The next query uses the new value.

Here is one session that sets both kinds. These are the orders.

spark.conf.set("spark.sql.shuffle.partitions", "8")   # runtime: works now

(orders.filter(F.col("status") == "paid")
       .groupBy("customer_id")
       .agg(F.sum("amount").alias("total"))
       .show())

spark.conf.set("spark.executor.memory", "40g")         # launch setting: error

The last line is too late. In Spark 3 it raises an error, an AnalysisException that says Spark cannot change this setting. The executors keep their size.

This table shows where to set each kind of value.

Put resource sizing in spark-submit or the cluster config. Tune queries in code.

You size a job from two things: the shape of your cluster, and the size of your data. Shape means the number of nodes (machines), and the cores and memory on each node.

First the cluster side. Two numbers here are guidelines that many teams use, not laws. Adjust them after you read the Spark UI.

Follow these five steps.

1. Reserve for the machine. Leave 1 core and about 1 GB on each node for the operating system and its background programs.
2. Choose cores per executor. A common guideline is about 5. Many more cores can slow reads and writes to HDFS, the Hadoop file storage. Garbage collection is when Java stops to free unused memory. A big executor pauses longer for it.
3. Count executors per node. Divide the usable cores by the cores per executor.
4. Work out memory. Divide the usable memory by the executors per node. Each slot must hold the executor memory and the overhead. Overhead is extra memory outside the Java heap: max(384 MB, 10% of executor memory).
5. On YARN, leave one executor for the application master. This is the process that asks YARN for executors. So total executors = nodes × executors per node − 1.

The table shows what goes wrong when a choice is too low or too high.

The data side comes later. It sets the number of partitions, and so how many rounds of tasks your cores must run.

Now we run the steps on a made-up cluster. It has 10 worker nodes. Each node has 16 cores and 64 GB of memory.

The table does the arithmetic one step at a time. Then the command shows the result.

spark-submit \
  --conf spark.executor.cores=5 \
  --conf spark.executor.memory=19g \
  --conf spark.executor.instances=29 \
  app.py

Now check the memory for each task. You know the model from "Memory Management & Tungsten". Follow these four lines.

1. 19 GB is 19 × 1,024 = 19,456 MB.
2. Take away the 300 MB that Spark reserves: 19,456 − 300 = 19,156 MB.
3. Multiply by spark.memory.fraction, 0.6: about 11,494 MB. Running tasks and cached data share this pool.
4. Five tasks run at once. With nothing cached, each task gets between 11,494 ÷ 10 and 11,494 ÷ 5: about 1,150 to 2,300 MB.

A partition of 100 to 200 MB fits in 1,150 MB. But data in memory is often bigger than the same data on disk, so keep a wide margin.

This is a starting point. Run the job, then check spill, garbage collection time and idle cores in the Spark UI.

The setting spark.sql.shuffle.partitions decides how many partitions a shuffle makes. The default is 200, for every job, big or small.

A common guideline is 100 to 200 MB of data in each shuffle partition. So the number of partitions is the shuffle data size divided by that target.

Take a shuffle that moves 100 GB, which is 102,400 MB. The table tries three settings.

Choose 800. On the 145-core cluster, the tasks run in waves. A wave is one round where every core runs one task.

Here 800 ÷ 145 = 5.5, so the stage needs 6 waves.

spark.conf.set("spark.sql.shuffle.partitions", "800")

AQE can help here. AQE (Adaptive Query Execution) is the feature that changes the plan while the job runs. It is on by default since Spark 3.2.

After a shuffle, AQE merges small partitions up to a target size. The target is at most spark.sql.adaptive.advisoryPartitionSizeInBytes, 64 MB by default.

So set the number a little high and let AQE merge. AQE only merges small partitions. It cannot fix a number that is too low.

The executor shape stays the same: 5 cores and 19 GB. Only the data changes. The data decides the number of partitions, and the partitions decide the number of waves.

We use the 145-core cluster again. Here "data" means the data that moves in the shuffle. Divide it by 128 MB to get partitions, then divide by 145 and round up to get waves.

Read the three rows one by one.

• 10 GB: 80 tasks fit in one wave, so 65 cores sit idle. Sixteen executors are enough (80 ÷ 5 = 16).
• 500 GB: 28 waves. If one task takes about a minute, the stage takes about 28 minutes. Add executors to shorten it.
• 10 TB: 565 waves is too long on this cluster. You need a much bigger cluster, or you accept a long job. Start with many shuffle partitions and let AQE merge small ones.

So far the number of executors is fixed. A fixed number wastes machines when a job has quiet stages.

Dynamic allocation lets Spark add executors when tasks wait, and remove them when they are idle. The setting is spark.dynamicAllocation.enabled. It is off (false) by default.

When Spark removes an executor, its shuffle files must stay available. Most setups use one of these two things.

• Shuffle tracking: Spark keeps an executor while other executors still need its shuffle files.
• The external shuffle service: a separate program that keeps the shuffle files.

Here is a command that turns it on with shuffle tracking.

spark-submit \
  --conf spark.dynamicAllocation.enabled=true \
  --conf spark.dynamicAllocation.shuffleTracking.enabled=true \
  --conf spark.dynamicAllocation.minExecutors=2 \
  --conf spark.dynamicAllocation.maxExecutors=29 \
  app.py

The table compares the two choices.

Four points to keep.

• Code beats spark-submit, and spark-submit beats spark-defaults.conf. Resource settings are fixed at launch.
• Size executors from the cluster. Use about 5 cores, reserve 1 core and 1 GB per node, split the memory, and keep 10% for overhead.
• Choose shuffle partitions from the data: shuffle size ÷ 100 to 200 MB. Let AQE merge small ones.
• Every number is a starting point. Check the Spark UI and adjust.

Answer these without looking back.

1. A flag says 400 and the code says 8. Which value does Spark use?
2. Why does spark.conf.set("spark.executor.memory", "40g") not help after the session starts?
3. A node has 32 cores and 128 GB. After the reserve, how many 5-core executors fit?
4. A shuffle moves 100 GB. How many partitions give about 128 MB each?
5. What two things can keep shuffle files when dynamic allocation removes an executor?