In our earlier put up Enhance operational efficiencies of Apache Iceberg tables constructed on Amazon S3 information lakes, we mentioned how one can implement options to enhance operational efficiencies of your Amazon Easy Storage Service (Amazon S3) information lake that’s utilizing the Apache Iceberg open desk format and working on the Amazon EMR massive information platform. Iceberg tables retailer metadata in manifest recordsdata. Because the variety of information recordsdata enhance, the quantity of metadata saved in these manifest recordsdata additionally will increase, resulting in longer question planning time. The question runtime additionally will increase as a result of it’s proportional to the variety of information or metadata file learn operations. Compaction is the method of mixing these small information and metadata recordsdata to enhance efficiency and scale back value. Compaction additionally removes deleting recordsdata by making use of deletes and rewriting a brand new file with out deleting data. At present, Iceberg offers a compaction utility that compacts small recordsdata at a desk or partition stage. However this strategy requires you to implement the compaction job utilizing your most well-liked job scheduler or manually triggering the compaction job.

On this put up, we focus on the brand new Iceberg characteristic that you should utilize to robotically compact small recordsdata whereas writing information into Iceberg tables utilizing Spark on Amazon EMR or Amazon Athena.

Use circumstances for processing small recordsdata

Streaming functions are liable to creating a lot of small recordsdata, which might negatively impression the efficiency of subsequent processing instances. For instance, take into account a important Web of Issues (IoT) sensor from a chilly storage facility that’s constantly sending temperature and well being information into an S3 information lake for downstream information processing and triggering actions like emergency upkeep. Techniques of this nature generate an enormous variety of small objects and want consideration to compact them to a extra optimum measurement for quicker studying, comparable to 128 MB, 256 MB, or 512 MB. On this put up, we present you a streaming sensor information use case with a lot of small recordsdata and the mitigation steps utilizing the Iceberg open desk format. For extra data on streaming functions on AWS, discuss with Actual-time Information Streaming and Analytics.

Resolution overview

To compact the small recordsdata for improved efficiency, on this instance, Amazon EMR triggers a compaction job after the write commit as a post-commit hook when outlined thresholds (for instance, variety of commits) are met. By default, Amazon EMR waits for 10 commits to set off the post-commit hook compaction utility.

This Iceberg event-based desk administration characteristic helps you to monitor desk actions throughout writes to make higher selections about easy methods to handle every desk in another way based mostly on occasions. As of this writing, solely the optimize-data optimization is supported. To be taught extra in regards to the obtainable optimize information executors and catalog properties, discuss with the README file within the GitHub repo.

To make use of the characteristic, you should utilize the iceberg-aws-event-based-table-management supply code and supply the constructed JAR within the engine’s class-path. The next bootstrap motion can place the JAR within the engine’s class-path:

sudo aws s3 cp s3://<path>/iceberg-aws-event-based-table-management-0.1.jar /usr/lib/spark/jars/

Notice that the Iceberg AWS event-based desk administration characteristic works with Iceberg v1.2.0 and above (obtainable from Amazon EMR 6.11.0).

In some use circumstances, you could wish to run the event-based compaction jobs in a unique EMR cluster with a purpose to keep away from any impression to the ETL jobs working of their present EMR cluster. You may get the metadata, together with the cluster ID of your present ETL workflows, from the /mnt/var/lib/information/job-flow.json file after which use a unique cluster to course of the event-based compactions.

The pocket book examples proven within the following sections are additionally obtainable within the aws-samples GitHub repo.

Prerequisite

For this efficiency comparability train between a Spark exterior desk and an Iceberg desk and Iceberg with compaction, we generate a major variety of small recordsdata in Parquet format and retailer them in an S3 bucket. We used the Amazon Kinesis Information Generator (KDG) instrument to generate pattern sensor information data utilizing the next template:

{"sensorId": {{random.quantity(5000)}},
 "currentTemperature": {{random.quantity(
        {
            "min":10,
            "max":150
        }
  )}},
 "standing": "{{random.arrayElement(
        ["OK","FAIL","WARN"]
    )}}",
 "date_ts": "{{date.now("YYYY-MM-DD HH:mm:ss")}}"
}

We configured an Amazon Kinesis Information Firehose supply stream and despatched the generated information right into a staging S3 bucket. Then we ran an AWS Glue extract, rework, and cargo (ETL) job to transform the JSON recordsdata into Parquet format. For our testing, we generated about 58,176 small objects with complete measurement of two GB.

For working the Amazon EMR exams, we used Amazon EMR model emr-6.11.0 with Spark 3.3.2, and JupyterEnterpriseGateway 2.6.0. The cluster used had one main node (r5.2xlarge) and two core nodes (r5.xlarge). We used a bootstrap motion throughout cluster creation to allow event-based desk administration:

sudo aws s3 cp s3://<path>/iceberg-aws-event-based-table-management-0.1.jar /usr/lib/spark/jars/

Additionally, discuss with our steerage on easy methods to use an Iceberg cluster with Spark, which is a prerequisite for this train.

As a part of the train, we see new steps are being added to the EMR cluster to set off the compaction jobs. To allow including new steps to the working cluster, we add the elasticmapreduce:AddJobFlowSteps motion to the cluster’s default position, EMR_EC2_DefaultRole, as a prerequisite.

Efficiency of Iceberg reads with the compaction utility on Amazon EMR

Within the following steps, we show easy methods to use the compaction utility and what efficiency advantages you may obtain. We use an EMR pocket book to show the advantages of the compaction utility. For directions to arrange an EMR pocket book, discuss with Amazon EMR Studio overview.

First, you configure your Spark session utilizing the %%configure magic command. We use the Hive catalog for Iceberg tables.

1. Earlier than you run the next step, create an Amazon S3 bucket in your AWS account known as <your-iceberg-storage-blog>. To examine easy methods to create an Amazon S3 bucket, comply with the directions given right here. Replace the your-iceberg-storage-blog bucket title within the following configuration with the precise bucket title you created to check this instance:

   %%configure -f
   {
   "conf":{
       "spark.sql.extensions":"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
       "spark.sql.catalog.dev":"org.apache.iceberg.spark.SparkCatalog",
       "spark.sql.catalog.dev.catalog-impl":"org.apache.iceberg.aws.glue.GlueCatalog",
       "spark.sql.catalog.dev.io-impl":"org.apache.iceberg.aws.s3.S3FileIO",
       "spark.sql.catalog.dev.warehouse":"s3://<your-iceberg-storage-blog>/iceberg/"
       }
   }

2. Create a brand new database for the Iceberg desk within the AWS Glue Information Catalog named DB and supply the S3 URI specified within the Spark config as s3://<your-iceberg-storage-blog>/iceberg/db. Additionally, create one other Database named iceberg_db in Glue for the parquet tables. Comply with the directions given in Working with databases on the AWS Glue console to create your Glue databases. Then create a brand new Spark desk in Parquet format pointing to the bucket containing small objects in your AWS account. See the next code:

   spark.sql(""" CREATE TABLE iceberg_db.sensor_data_parquet_table (
       sensorid int,
       currenttemperature int,
       standing string,
       date_ts timestamp)
   USING parquet
   location 's3://<your-bucket-with-parquet-files>/'
   """)

3. Run an combination SQL to measure the efficiency of Spark SQL on the Parquet desk with 58,176 small objects:

   spark.sql(""" choose maxtemp, mintemp, avgtemp from
   (choose
   max(currenttemperature) as maxtemp,
   min(currenttemperature) as mintemp,
   avg(currenttemperature) as avgtemp
   from iceberg_db.sensor_data_parquet_table
   the place month(date_ts) between 2 and 10
   order by maxtemp, mintemp, avgtemp)""").present()

Within the following steps, we create a brand new Iceberg desk from the Spark/Parquet desk utilizing CTAS (Create Desk As Choose). Then we present how the automated compaction job may also help enhance question efficiency.

5. Create a brand new Iceberg desk utilizing CTAS from the sooner AWS Glue desk with the small recordsdata:

   spark.sql(""" CREATE TABLE dev.db.sensor_data_iceberg_format USING iceberg AS (SELECT * FROM iceberg_db.sensor_data_parquet_table)""")

6. Validate {that a} new Iceberg snapshot was created for the brand new desk:

   spark.sql(""" Choose * from dev.db.sensor_data_iceberg_format.snapshots restrict 5""").present()

We’ve confirmed that our S3 folder corresponds to the newly created Iceberg desk. It reveals that through the CTAS assertion, it added 1,879 objects within the new folder with a complete measurement of 1.3 GB. We will conclude that Iceberg did some optimization whereas loading information from the Parquet desk.

7. Now that you’ve information within the Iceberg desk, run the earlier aggregation SQL to examine the runtime:

   spark.sql(""" choose maxtemp, mintemp, avgtemp from
   (choose
   max(currenttemperature) as maxtemp,
   min(currenttemperature) as mintemp,
   avg(currenttemperature) as avgtemp
   from dev.db.sensor_data_iceberg_format
   the place month(date_ts) between 2 and 10
   order by maxtemp, mintemp, avgtemp)""").present()

The runtime for the previous question ran on the Iceberg desk with 1,879 objects in 1 minute, 39 seconds. There may be already some vital efficiency enchancment by changing the exterior Parquet desk to an Iceberg desk.

8. Now let’s add the configurations wanted to use the automated compaction of small recordsdata within the Iceberg tables. Notice the final 4 newly added configurations within the following assertion. The parameter optimize-data.commit-threshold means that the compaction will happen after the primary profitable commit. The default is 10 profitable commits to set off the compaction.

   %%configure -f
   {
   "conf":{
       "spark.sql.extensions":"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
       "spark.sql.catalog.dev":"org.apache.iceberg.spark.SparkCatalog",
       "spark.sql.catalog.dev.catalog-impl":"org.apache.iceberg.aws.glue.GlueCatalog",
       "spark.sql.catalog.dev.io-impl":"org.apache.iceberg.aws.s3.S3FileIO",
       "spark.sql.catalog.dev.warehouse":"s3://<your-iceberg-storage-blog>/iceberg/",
       "spark.sql.catalog.dev.metrics-reporter-impl":"org.apache.iceberg.aws.handle.AwsTableManagementMetricsEvaluator",
       "spark.sql.catalog.dev.optimize-data.impl":"org.apache.iceberg.aws.handle.EmrOnEc2OptimizeDataExecutor",
       "spark.sql.catalog.dev.optimize-data.emr.cluster-id":"j-1N8J5NZI0KEU3",
       "spark.sql.catalog.dev.optimize-data.commit-threshold":"1"
       }
   }

9. Run a fast sanity examine to substantiate that the configurations are working fantastic with Spark SQL.

10. 10. To activate the automated compaction course of, add a brand new document to the present Iceberg desk utilizing a Spark insert:

    spark.sql(""" Insert into dev.db.sensor_data_iceberg_format values(999123, 86, 'PASS', timestamp'2023-07-26 12:50:25') """)

11. Navigate to the Amazon EMR console to examine the cluster steps.

It’s best to see a brand new step added that goes from Pending to Operating and at last the Accomplished state. Each time the information within the Iceberg desk is up to date or inserted, based mostly on configuration optimize-data.commit-threshold, the optimize job will robotically set off to compact the underlying information.

12. Validate that the document insert was profitable.

13. Examine the snapshot desk to see {that a} new snapshot is created for the desk with the operation substitute.

For each profitable run of the background optimize job, a brand new entry will probably be added to the snapshot desk.

14. On the Amazon S3 console, navigate to the folder equivalent to the Iceberg desk and see that the information recordsdata are compacted.

In our case, it was compacted from the earlier smaller sizes to roughly 437 MB. The folder will nonetheless comprise the earlier smaller recordsdata for time journey except you subject an expire snapshot command to take away them.

15. Now you may run the identical combination question and document the efficiency after the compaction.

Abstract of Amazon EMR testing

The runtime for the previous aggregation question on the compacted Iceberg desk diminished to roughly 59 seconds from the earlier runtime of 1 minute, 39 seconds. That’s a few 40% enchancment. The extra small recordsdata you’ve in your supply bucket, the larger efficiency increase you may obtain with this post-hook compaction implementation. The examples proven on this weblog have been executed in a small Amazon EMR cluster with solely two core nodes (r5.xlarge). To enhance the efficiency of your Spark functions, Amazon EMR offers a number of optimization options you could implement to your manufacturing workloads.

Efficiency of Iceberg reads with the compaction utility on Athena

To handle the Iceberg desk based mostly on occasions, you can begin the Spark 3.3 SQL shell as proven within the following code. Guarantee that the athena:StartQueryExecution and athena:GetQueryExecution permission insurance policies are enabled.

spark-sql --conf spark.sql.catalog.my_catalog=org.apache.iceberg.spark.SparkCatalog 
          --conf spark.sql.catalog.my_catalog.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog 
          --conf spark.sql.catalog.my_catalog.warehouse=<s3-bucket> 
          --conf spark.sql.catalog.my_catalog.metrics-reporter-impl=org.apache.iceberg.aws.handle.AwsTableManagementMetricsEvaluator 
          --conf spark.sql.catalog.my_catalog.optimize-data.impl=org.apache.iceberg.aws.handle.AthenaOptimizeDataExecutor 
          --conf spark.sql.catalog.my_catalog.optimize-data.athena.output-bucket=<s3-bucket>

Clear up

After you full the take a look at, clear up your sources to keep away from any recurring prices:

1. Delete the S3 buckets that you simply created for this take a look at.
2. Delete the EMR cluster.
3. Cease and delete the EMR pocket book occasion.

Conclusion

On this put up, we confirmed how Iceberg event-based desk administration helps you to handle every desk in another way based mostly on occasions and compact small recordsdata to spice up utility efficiency. This event-based course of considerably reduces the operational overhead of utilizing the Iceberg rewrite_data_files process, which wants handbook or scheduled operation.

To be taught extra about Apache Iceberg and implement this open desk format to your transactional information lake use circumstances, discuss with the next sources:

In regards to the Authors

Avijit Goswami is a Principal Options Architect at AWS specialised in information and analytics. He helps AWS strategic clients in constructing high-performing, safe, and scalable information lake options on AWS utilizing AWS managed providers and open-source options. Outdoors of his work, Avijit likes to journey, hike, watch sports activities, and hearken to music.

Rajarshi Sarkar is a Software program Growth Engineer at Amazon EMR/Athena. He works on cutting-edge options of Amazon EMR/Athena and can be concerned in open-source initiatives comparable to Apache Iceberg and Trino. In his spare time, he likes to journey, watch films, and hang around with buddies.