How do I resolve the "No space left on device" error in an AWS Glue ETL job?

Short description

You might receive the following types of errors when you run an AWS Glue extract, transform, and load (ETL) job:

• Job aborted due to stage failure: Task 68 in stage 158.0 failed 4 times, most recent failure: Lost task 68.3 in stage 158.0 (TID 28820, 10.102.100.111, executor 17): org.apache.spark.memory.SparkOutOfMemoryError: error while calling spill() on org.apache.spark.shuffle.sort.ShuffleExternalSorter@55f6dabb: No space left on device

        -or-

• Job aborted due to stage failure: ResultStage 7 has failed the maximum allowable number of times: 4. Most recent failure reason: org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shuffle 2

Apache Spark uses local disk on AWS Glue workers to spill data from memory that exceeds the heap space defined by the spark.executor.memory configuration parameter.

Wide transformations such as groupByKey(), reduceByKey(), and join() can cause a shuffle. During the job’s sort or shuffle stage, Spark writes the intermediate data to a local disk before it can exchange that data between the different workers. At this point, you might get a “No space left on device“ or a MetadataFetchFailedException error. Spark throws this error when there isn’t enough disk space left on the executor and there's no recovery.

Resolution

These types of errors commonly occur when the processing job observes a significant skew in the dataset. This section lists some of the common monitoring and debugging anomalies and the ways that you can address them.

AWS Glue job metrics and Apache Spark UI are powerful tools for monitoring data skew in the Apache Spark executors. By monitoring the implementation timeline, the tool lets you easily identify any issues that may result in data skew. It helps you understand the behavior of each stage, task, job, and executor in detail.

Disaggregate compute and storage

This approach scales storage for large shuffles instead of writing the data to the AWS Glue workers’ local disk.

Use dedicated serverless storage: AWS Glue 2.0 or higher lets you use Amazon Simple Storage Service (Amazon S3) to store Spark shuffle and spill data.

AWS Glue 2.0 Use the following job parameters to use Amazon S3 shuffle in AWS Glue. For more information, see AWS Glue Spark shuffle plugin with Amazon S3.

• Key: --write-shuffle-files-to-s3
  Value: TRUE
• Key: --write-shuffle-spills-to-s3
  Value: TRUE
• Key: --conf
  Value: spark.shuffle.glue.s3ShuffleBucket=s3://custom_shuffle_bucket
  Note: The optional flag spark.shuffle.glue.s3ShuffleBucket specifies the Amazon S3 bucket that you write the shuffle files to. Replace custom_shuffle_bucket with the name of your S3 bucket.

AWS Glue 3.0/4.0 Use following job parameters to use cloud shuffle storage plugin for Apache Spark. For more information, see Cloud Shuffle Storage Plugin for Apache Spark.

• Key: --write-shuffle-files-to-s3
  Value: TRUE
• Key: --conf
  Value: spark.shuffle.storage.path=s3://custom_shuffle_bucket
  Note: The optional flag spark.shuffle.storage.path specifies the Amazon S3 bucket that you write the shuffle files to. Replace custom_shuffle_bucket with the name of your S3 bucket.

Scaling out

Scaling out refers to either increasing the number of workers through horizontal scaling or upgrading the worker type through vertical scaling. However, scaling out might not always work, especially if your data is heavily skewed on a few keys. To address data skew, consider modifying the Apache Spark application logic by implementing the salting technique.

Reduce and filter input data

Pre-filter input data as much as possible to minimize data shuffling and network usage during wide operations. Use the following techniques for effective data filtering:

• AWS Glue push down predicates
• AWS Glue workload partitioning with bounded execution
• AWS Glue exclusions for filtering

Broadcast small tables

Joining tables in Apache Spark can trigger data shuffling and movement of massive amounts of data between the executors on different workers. This can cause the system to run out of memory and spill the data onto the worker's disk. To minimize network overhead, Spark supports the use of smaller tables for broadcast. These tables are no larger than tens of MBs (megabytes) and prevent data partitioning and shuffling. Use a smaller table for broadcast by providing a hint to Spark. For more information, see Broadcast small tables in the AWS blog Optimize memory management in AWS Glue.

Use AQE

Adaptive Query Execution (AQE) from Databricks is an optimization technique in Spark SQL. It uses runtime statistics to choose the most efficient query implementation plan to resolve data skew and dynamic shuffle partitions at each stage. AQE is available by default in AWS Glue 3.0/4.0.

AQE performs the following functions:

• Dynamically coalesces shuffle partitions
  Automatically coalesces partitions, if needed, after each query stage and solves partition management problems. For more information, see Dynamically coalescing shuffle partitions in Adaptive Query Execution: Speeding Up Spark SQL at Runtime on the Databricks website.
  –Too few partitions: Automatically splits partitions with excess data that cause a spill to the local disk.
  –Too many partitions: Automatically combines partitions, as needed. Partitions are combined when the data size of each partition is small and requires small network data fetches to read the shuffle blocks.

• Dynamically switches join strategies
  Converts Sort-Merge to Broadcast Hash Join, based on the table size statistics in each stage.

• Dynamically optimizes skew joins
  Automatically detects and optimizes skew join from shuffle file statistics. AQE splits the skewed partitions into smaller subpartitions. Then, it joins these smaller partitions to the corresponding partition in the other node. For more information, see Dynamically optimizing skew joins in Adaptive Query Execution: Speeding Up Spark SQL at Runtime on the Databricks website.

Configure AQE

Use the following job parameters to turn on AQE:

• Key: --conf
  Value: spark.sql.adaptive.enabled=true --conf spark.sql.adaptive.coalescePartitions.enabled=true --conf spark.sql.adaptive.skewJoin.enabled=true