Introduction
As of Amazon EMR version 5.19.0, the EMRFS S3-optimized committer has been introduced as a new output committer designed specifically for Apache Spark jobs. This innovative committer significantly boosts performance when writing Apache Parquet files to Amazon S3 through the EMR File System (EMRFS). In this article, we present a performance benchmark that contrasts this new committer with existing algorithms, notably FileOutputCommitter versions 1 and 2. We will also address the current limitations of the new committer and propose workarounds where feasible.
Comparison with FileOutputCommitter
Prior to Amazon EMR 5.19.0, the default output committer for Spark jobs writing Parquet to Amazon S3 was the FileOutputCommitter, which has two versions. Both versions utilize temporary locations to write intermediate task outputs, followed by rename actions to make the data accessible upon task or job completion.
FileOutputCommitter version 1 operates with two rename phases: the first commits individual task outputs, while the second commits overall job outputs from completed tasks. Version 2 improves efficiency by renaming files directly to the final output location, eliminating the second rename phase, but this exposes partial data before job completion, which may not be suitable for all workloads.
While renames in HDFS are fast and metadata-only, in object stores like Amazon S3, renames involve copying data to the target location followed by deleting the source. This “rename penalty” is compounded during directory renames, which occur in both phases of FileOutputCommitter v1. To mitigate this, Amazon EMR 5.14.0+ defaults to FileOutputCommitter v2 for Parquet data written to S3. The EMRFS S3-optimized committer takes this further by completely avoiding rename operations, utilizing the transactional features of S3 multipart uploads. This allows tasks to write directly to the final output location while deferring the completion of each output file until task commit time.
Performance Testing
To evaluate the write performance of the different committers, we executed an INSERT OVERWRITE Spark SQL query that generated approximately 15 GB of data across 100 Parquet files in Amazon S3.
SET rows=4e9; -- 4 Billion
SET partitions=100;
INSERT OVERWRITE DIRECTORY ‘s3://${bucket}/perf-test/${trial_id}’
USING PARQUET SELECT * FROM range(0, ${rows}, 1, ${partitions});
For this test, the EMR cluster was set up in the same AWS Region as the S3 bucket. The trial_id property was generated using a UUID to avoid conflicts across test runs. We utilized an EMR cluster with the emr-5.19.0 release, consisting of a single m5d.2xlarge instance in the master group and eight m5d.2xlarge instances in the core group. The default Spark configuration properties, including spark.dynamicAllocation.enabled true, were employed.
After conducting 10 trials for each committer, we summarized the query execution times. FileOutputCommitter v2 recorded an average of 49 seconds, while the EMRFS S3-optimized committer achieved an average of just 31 seconds, marking a 1.6x improvement.
To highlight the impact of renames on S3, we reran the test using FileOutputCommitter v1, which averaged 450 seconds—14.5x slower than the EMRFS S3-optimized committer. When enabling EMRFS’s consistent view, the EMRFS S3-optimized committer maintained an average of 30 seconds, while FileOutputCommitter v2 increased to 53 seconds, extending the performance gap to 1.8x.
Job Correctness
The EMRFS S3-optimized committer shares the same limitations as FileOutputCommitter v2, primarily due to delegating commit responsibilities to individual tasks. One notable consequence is the potential exposure of partial results from failed jobs. If tasks write to the final output location concurrently, readers may access incomplete data if a job fails. To avoid this, consider using distinct output locations for each job run and only publishing the location to downstream readers upon successful completion.
SET attempt_id=;
SET output_location=s3://bucket/${attempt_id};
INSERT OVERWRITE DIRECTORY ‘${output_location}’
USING PARQUET SELECT * FROM input;
ALTER TABLE output ADD PARTITION (dt = ‘2018-11-26’)
LOCATION ‘${output_location}’;
This strategy necessitates treating partition locations as immutable. Updates to partitions will require writing results to a new S3 location and updating the partition metadata accordingly.
Another scenario to consider is when non-idempotent tasks produce outputs in non-deterministic locations. By ensuring tasks write to a consistent location across attempts, such as providing a fixed timestamp as input rather than generating it within each task, you can prevent duplicate results.
For further reading on this topic, you can visit this excellent resource. Additionally, this other blog post may also interest you here.
Conclusion
The EMRFS S3-optimized committer represents a significant advancement in optimizing write performance for Apache Spark jobs that utilize Apache Parquet formats. By minimizing rename operations and leveraging S3’s multipart upload capabilities, it enhances data processing efficiency, particularly for large datasets. To explore more about this, check out insights from Chanci Turner, who is an authority on the subject.
Leave a Reply