In the era of data, organizations are increasingly using data lakes to store and analyze vast amounts of structured and unstructured data. Data lakes provide a centralized repository for data from various sources, enabling organizations to unlock valuable insights and drive data-driven decision-making. However, as data volumes continue to grow, optimizing data layout and organization becomes crucial for efficient querying and analysis.
One of the key challenges in data lakes is the potential for slow query performance, especially when dealing with large datasets. This can be attributed to factors such as inefficient data layout, resulting in excessive data scanning and inefficient use of compute resources. To address this challenge, common practices like partitioning and bucketing can significantly improve query performance and reduce computation costs.
Partitioning is a technique that divides a large dataset into smaller, more manageable parts based on specific criteria, such as date, region, or product category. By partitioning data, downstream analytical queries can skip irrelevant partitions, reducing the amount of data that needs to be scanned and processed. You can use partition columns in the WHERE clause in queries to scan only the specific partitions that your query needs. This can lead to faster query runtimes and more efficient resource utilization. It especially works well when columns with low cardinality are chosen as the key.
What if you have a high cardinality column that you sometimes need to filter by VIP customers? Each customer is usually identified with an ID, which can be millions. Partitioning isn’t suitable for such high cardinality columns because you end up with small files, slow partition filtering, and high Amazon Simple Storage Service (Amazon S3) API cost (one S3 prefix is created per value of partition column). Although you can use partitioning with a natural key such as city or state to narrow down your dataset to some degree, it is still necessary to query across date-based partitions if your data is time series.
This is where bucketing comes into play. Bucketing makes sure that all rows with the same values of one or more columns end up in the same file. Instead of one file per value, like partitioning, a hash function is used to distribute values evenly across a fixed number of files. By organizing data this way, you can perform efficient filtering, because only the relevant buckets need to be processed, further reducing computational overhead.
There are multiple options for implementing bucketing on AWS. One approach is to use the Amazon Athena CREATE TABLE AS SELECT (CTAS) statement, which allows you to create a bucketed table directly from a query. Alternatively, you can use AWS Glue for Apache Spark, which provides built-in support for bucketing configurations during the data transformation process. AWS Glue allows you to define bucketing parameters, such as the number of buckets and the columns to bucket on, providing an optimized data layout for efficient querying with Athena.
In this post, we discuss how to implement bucketing on AWS data lakes, including using Athena CTAS statement and AWS Glue for Apache Spark. We also cover bucketing for Apache Iceberg tables.
Example use case In this post, you use a public dataset, the NOAA Integrated Surface Database. Data analysts run one-time queries for data during the past 5 years through Athena. Most of the queries are for specific stations with specific report types. The queries need to complete in 10 seconds, and the cost needs to be optimized carefully. In this scenario, you’re a data engineer responsible for optimizing query performance and cost.
For example, if an analyst wants to retrieve data for a specific station (for example, station ID 123456) with a particular report type (for example, CRN01), the query might look like the following query:
SELECT station, report_type, columnA, columnB, ...FROM table_nameWHEREreport_type = 'CRN01'AND station = '123456'
In the case of the NOAA Integrated Surface Database, the station_id column is likely to have a high cardinality, with numerous unique station identifiers. On the other hand, the report_type column may have a relatively low cardinality, with a limited set of report types. Given this scenario, it would be a good idea to partition the data by report_type and bucket it by station_id.
With this partitioning and bucketing strategy, Athena can first eliminate partitions for irrelevant report types, and then scan only the buckets within the relevant partition that match the specified station ID, significantly reducing the amount of data processed and accelerating query runtimes. This approach not only meets the query performance requirement, but also helps optimize costs by minimizing the amount of data scanned and billed for each query.
In this post, we examine how query performance is affected by data layout, in particular, bucketing. We also compare three different ways to achieve bucketing. The following table represents conditions for the tables to be created.
| . | noaa_remote_original | athena_non_bucketed | athena_bucketed | glue_bucketed | athena_bucketed_iceberg | | Format | CSV | Parquet | Parquet | Parquet | Parquet | | Compression | n/a | Snappy | Snappy | Snappy | Snappy | | Created via | n/a | Athena CTAS | Athena CTAS | Glue ETL | Athena CTAS with Iceberg | | Engine | n/a | Trino | Trino | Apache Spark | Apache Iceberg | | Is partitioned? | Yes but with different way | Yes | Yes | Yes | Yes | | Is bucketed? | No | No | Yes | Yes | Yes |
noaa_remote_original is partitioned by the year column, but not by the report_type column. This row represents if the table is partitioned by the actual columns that are used in the queries.
Baseline table For this post, you create several tables with different conditions: some without bucketing and some with bucketing, to showcase the performance characteristics of bucketing. First, let’s create an original table using the NOAA data. In subsequent steps, you ingest data from this table to create test tables.
There are multiple ways to define a table definition: running DDL, an AWS Glue crawler, the AWS Glue Data Catalog API, and so on. In this step, you run DDL via the Athena console.
Complete the following steps to create the "bucketing_blog"."noaa_remote_original" table in the Data Catalog:
-- Create Glue databaseCREATE DATABASE bucketing_blog;bucketing_blog to set the current database.-- Create original tableCREATE EXTERNAL TABLE `bucketing_blog`.`noaa_remote_original`( `station` STRING, `date` STRING, `source` STRING, `latitude` STRING, `longitude` STRING, `elevation` STRING, `name` STRING, `report_type` STRING, `call_sign` STRING, `quality_control` STRING, `wnd` STRING, `cig` STRING, `vis` STRING, `tmp` STRING, `dew` STRING, `slp` STRING, `aj1` STRING, `gf1` STRING, `mw1` STRING)PARTITIONED BY ( year STRING)ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.OpenCSVSerde' WITH SERDEPROPERTIES ( 'escapeChar'='\\', 'quoteChar'='\"', 'separatorChar'=',') STORED AS INPUTFORMAT 'org.apache.hadoop.mapred.TextInputFormat' OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'LOCATION 's3://noaa-global-hourly-pds/'TBLPROPERTIES ( 'skip.header.line.count'='1')Because the source data has quoted fields, we use OpenCSVSerde instead of the default LazySimpleSerde.
These CSV files have a header row, which we tell Athena to skip by adding skip.header.line.count and setting the value to 1.
For more details, refer to OpenCSVSerDe for processing CSV.
-- Load partitionsALTER TABLE `bucketing_blog`.`noaa_remote_original` ADD PARTITION (year = '2024') LOCATION 's3://noaa-global-hourly-pds/2024/' PARTITION (year = '2023') LOCATION 's3://noaa-global-hourly-pds/2023/' PARTITION (year = '2022') LOCATION 's3://noaa-global-hourly-pds/2022/' PARTITION (year = '2021') LOCATION 's3://noaa-global-hourly-pds/2021/' PARTITION (year = '2020') LOCATION 's3://noaa-global-hourly-pds/2020/';-- Check data SELECT * FROM "bucketing_blog"."noaa_remote_original" LIMIT 10;Now you’re ready to start querying the original table to examine the baseline performance.
CRN05:-- BaselineSELECT station, report_type, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, tmpFROM "bucketing_blog"."noaa_remote_original"WHERE report_type = 'CRN05' AND ( station = '99999904237' OR station = '99999953132' OR station = '99999903061' OR station = '99999963856' OR station = '99999994644' );We ran this query 10 times. The average query runtime for 10 queries is 27.6 seconds, which is far longer than our target of 10 seconds, and 155.75 GB data is scanned to return 1.65 million records. This is the baseline performance of the original raw table. It’s time to start optimizing data layout from this baseline.
Next, you create tables with different conditions from the original: one without bucketing and one with bucketing, and compare them.
Optimize data layout using Athena CTAS In this section, we use an Athena CTAS query to optimize data layout and its format.
First, let’s create a table with partitioning but without bucketing. The new table is partitioned by the column report_type because most of expected queries use this column in the WHERE clause, and objects are stored as Parquet with Snappy compression.
--CTAS, non-bucketedCREATE TABLE "bucketing_blog"."athena_non_bucketed"WITH ( external_location = 's3://<your-s3-location>/athena-non-bucketed/', partitioned_by = ARRAY['report_type'], format = 'PARQUET', write_compression = 'SNAPPY')ASSELECT station, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, vis, tmp, dew, slp, aj1, gf1, mw1, report_typeFROM "bucketing_blog"."noaa_remote_original";Your data should look like the following screenshots.
There are 30 files under the partition.
Next, you create a table with Hive style bucketing. The number of buckets needs to be carefully tuned through experiments for your own use case. Generally speaking, the more buckets you have, the smaller the granularity, which might result in better performance. On the other hand, too many small files may introduce inefficiency in query planning and processing. Also, bucketing only works if you are querying a few values of the bucketing key. The more values you add to your query, the more likely that you will end up reading all buckets.
The following is the baseline query to optimize:
-- BaselineSELECT station, report_type, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, tmpFROM "bucketing_blog"."noaa_remote_original"WHERE report_type = 'CRN05' AND ( station = '99999904237' OR station = '99999953132' OR station = '99999903061' OR station = '99999963856' OR station = '99999994644' );
In this example, the table is going to be bucketed into 16 buckets by a high-cardinality column (station), which is supposed to be used for the WHERE clause in the query. All other conditions remain the same. The baseline query has five values in the station ID, and you expect queries to have around that number at most, which is less enough than the number of buckets, so 16 should work well. It is possible to specify a larger number of buckets, but CTAS can’t be used if the total number of partitions exceeds 100.
-- CTAS, Hive-bucketedCREATE TABLE "bucketing_blog"."athena_bucketed"WITH ( external_location = 's3://<your-s3-location>/athena-bucketed/', partitioned_by = ARRAY['report_type'], bucketed_by = ARRAY['station'], bucket_count = 16, format = 'PARQUET', write_compression = 'SNAPPY')ASSELECT station, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, vis, tmp, dew, slp, aj1, gf1, mw1, report_typeFROM "bucketing_blog"."noaa_remote_original";The query creates S3 objects organized as shown in the following screenshots.
The table-level layout looks exactly the same between athena_non_bucketed and athena_bucketed: there are 13 partitions in each table. The difference is the number of objects under the partitions. There are 16 objects (buckets) per partition, of roughly 10–25 MB each in this case. The number of buckets is constant at the specified value regardless of the amount of data, but the bucket size depends on the amount of data.
Now you’re ready to query against each table to evaluate query performance. The query will select records with five specific stations and report type CRN05 for the past 5 years. Although you can’t see which data of a specific station is located in which bucket, it has been calculated and located correctly by Athena.
-- No bucketing SELECT station, report_type, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, tmpFROM "bucketing_blog"."athena_non_bucketed"WHERE report_type = 'CRN05' AND ( station = '99999904237' OR station = '99999953132' OR station = '99999903061' OR station = '99999963856' OR station = '99999994644' );We ran this query 10 times. The average runtime of the 10 queries is 10.95 seconds, and 358 MB of data is scanned to return 2.21 million records. Both the runtime and scan size have been significantly decreased because you’ve partitioned the data, and can now read only one partition where 12 partitions of 13 are skipped. In addition, the amount of data scanned has gone down from 206 GB to 360 MB, which is a reduction of 99.8%. This is not just due to the partitioning, but also due to the change of its format to Parquet and compression with Snappy.
-- Hive bucketingSELECT station, report_type, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, tmpFROM "bucketing_blog"."athena_bucketed"WHERE report_type = 'CRN05' AND ( station = '99999904237' OR station = '99999953132' OR station = '99999903061' OR station = '99999963856' OR station = '99999994644' );We ran this query 10 times. The average runtime of the 10 queries is 7.82 seconds, and 69 MB of data is scanned to return 2.21 million records. This means a reduction of average runtime from 10.95 to 7.82 seconds (-29%), and a dramatic reduction of data scanned from 358 MB to 69 MB (-81%) to return the same number of records compared with the non-bucketed table. In this case, both runtime and data scanned were improved by bucketing. This means bucketing contributed not only to performance but also to cost reduction.
Considerations As stated earlier, size your bucket carefully to maximize performance of your query. Bucketing only works if you are querying a few values of the bucketing key. Consider creating more buckets than the number of values expected in the actual query.
Additionally, an Athena CTAS query is limited to create up to 100 partitions at one time. If you need a large number of partitions, you may want to use AWS Glue extract, transform, and load (ETL), although there is a workaround to split into multiple SQL statements.
Optimize data layout using AWS Glue ETL Apache Spark is an open source distributed processing framework that enables flexible ETL with PySpark, Scala, and Spark SQL. It allows you to partition and bucket your data based on your requirements. Spark has several tuning options to accelerate jobs. You can effortlessly automate and monitor Spark jobs. In this section, we use AWS Glue ETL jobs to run Spark code to optimize data layout.
Unlike Athena bucketing, AWS Glue ETL uses Spark-based bucketing as a bucketing algorithm. All you need to do is add the following table property onto the table: bucketing_format = 'spark'. For details about this table property, see Partitioning and bucketing in Athena.
Complete the following steps to create a table with bucketing through AWS Glue ETL:
bucketing_blog.noaa_remote_original.ToS3WithBucketing.def ToS3WithBucketing (glueContext, dfc) -> DynamicFrameCollection: # Convert DynamicFrame to DataFrame df = dfc.select(list(dfc.keys())[0]).toDF() # Write to S3 with bucketing and partitioning df.repartition(1, "report_type") \ .write.option("path", "s3://<your-s3-location>/glue-bucketed/") \ .mode("overwrite") \ .partitionBy("report_type") \ .bucketBy(16, "station") \ .format("parquet") \ .option("compression", "snappy") \ .saveAsTable("bucketing_blog.glue_bucketed")The following screenshot shows the job created using AWS Glue Studio to generate a table and data.
Each node represents the following:
noaa_remote_original table from the Data CatalogThe job has been successfully authored in the visual editor.
After these steps, the table glue_bucketed. has been created.
glue_bucketed.bucketing_format and value spark.Now it’s time to query the tables.
-- Spark bucketingSELECT station, report_type, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, tmpFROM "bucketing_blog"."glue_bucketed"WHERE report_type = 'CRN05' AND ( station = '99999904237' OR station = '99999953132' OR station = '99999903061' OR station = '99999963856' OR station = '99999994644' );We ran the query 10 times. The average runtime of the 10 queries is 7.09 seconds, and 88 MB of data is scanned to return 2.21 million records. In this case, both the runtime and data scanned were improved by bucketing. This means bucketing contributed not only to performance but also to cost reduction.
The reason for the larger bytes scanned compared to the Athena CTAS example is that the values were distributed differently in this table. In the AWS Glue bucketed table, the values were distributed over five files. In the Athena CTAS bucketed table, the values were distributed over four files. Remember that rows are distributed into buckets using a hash function. The Spark bucketing algorithm uses a different hash function than Hive, and in this case, it resulted in a different distribution across the files.
Considerations Glue DynamicFrame does not support bucketing natively. You need to use Spark DataFrame instead of DynamicFrame to bucket tables.
For information about fine-tuning AWS Glue ETL performance, refer to Best practices for performance tuning AWS Glue for Apache Spark jobs.
Optimize Iceberg data layout with hidden partitioning Apache Iceberg is a high-performance open table format for huge analytic tables, bringing the reliability and simplicity of SQL tables to big data. Recently, there has been a huge demand to use Apache Iceberg tables to achieve advanced capabilities like ACID transaction, time travel query, and more.
In Iceberg, bucketing works differently than the Hive table method we’ve seen so far. In Iceberg, bucketing is a subset of partitioning, and can be applied using the bucket partition transform. The way you use it and the end result is similar to bucketing in Hive tables. For more details about Iceberg bucket transforms, refer to Bucket Transform Details.
Complete the following steps:
-- CTAS, Iceberg-bucketedCREATE TABLE "bucketing_blog"."athena_bucketed_iceberg"WITH (table_type = 'ICEBERG', location = 's3://<your-s3-location>/athena-bucketed-iceberg/', is_external = false, partitioning = ARRAY['report_type', 'bucket(station, 16)'], format = 'PARQUET', write_compression = 'SNAPPY') ASSELECT station, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, vis, tmp, dew, slp, aj1, gf1, mw1, report_typeFROM "bucketing_blog"."noaa_remote_original";Your data should look like the following screenshot.
There are two folders: data and metadata. Drill down to data.
You see random prefixes under the data folder. Choose the first one to view its details.
You see the top-level partition based on the report_type column. Drill down to the next level.
You see the second-level partition, bucketed with the station column.
The Parquet data files exist under these folders.
-- Iceberg bucketingSELECT station, report_type, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, tmpFROM "bucketing_blog"."athena_bucketed_iceberg"WHERE report_type = 'CRN05' AND ( station = '99999904237' OR station = '99999953132' OR station = '99999903061' OR station = '99999963856' OR station = '99999994644' );With the Iceberg-bucketed table, the average runtime of the 10 queries is 8.03 seconds, and 148 MB of data is scanned to return 2.21 million records. This is less efficient than bucketing with AWS Glue or Athena, but considering the benefits of Iceberg’s various features, it is within an acceptable range.
Results The following table summarizes all the results.
| . | noaa_remote_original | athena_non_bucketed | athena_bucketed | glue_bucketed | athena_bucketed_iceberg | | Format | CSV | Parquet | Parquet | Parquet | Iceberg (Parquet) | | Compression | n/a | Snappy | Snappy | Snappy | Snappy | | Created via | n/a | Athena CTAS | Athena CTAS | Glue ETL | Athena CTAS with Iceberg | | Engine | n/a | Trino | Trino | Apache Spark | Apache Iceberg | | Table size (GB) | 155.8 | 5.0 | 5.0 | 5.8 | 5.0 | | The number of S3 Objects | 53360 | 376 | 192 | 192 | 195 | | Is partitioned? | Yes but with different way | Yes | Yes | Yes | Yes | | Is bucketed? | No | No | Yes | Yes | Yes | | Bucketing format | n/a | n/a | Hive | Spark | Iceberg | | Number of buckets | n/a | n/a | 16 | 16 | 16 | | Average runtime (sec) | 29.178 | 10.950 | 7.815 | 7.089 | 8.030 | | Scanned size (MB) | 206640.0 | 358.6 | 69.1 | 87.8 | 147.7 |
With athena_bucketed, glue_bucketed, and athena_bucketed_iceberg, you were able to meet the latency goal of 10 seconds. With bucketing, you saw a 25–40% reduction in runtime and a 60–85% reduction in scan size, which can contribute to both latency and cost optimization.
As you can see from the result, although partitioning contributes significantly to reduce both runtime and scan size, bucketing can also contribute to reduce them further.
Athena CTAS is straightforward and fast enough to complete the bucketing process. AWS Glue ETL is more flexible and scalable to achieve advanced use cases. You can choose either method based on your requirement and use case, because you can take advantage of bucketing through either option.
Conclusion In this post, we demonstrated how to optimize your table data layout with partitioning and bucketing through Athena CTAS and AWS Glue ETL. We showed that bucketing contributes to accelerating query latency and reducing scan size to further optimize costs. We also discussed bucketing for Iceberg tables through hidden partitioning.
Bucketing just one technique to optimize data layout by reducing data scan. For optimizing your entire data layout, we recommend considering other options like partitioning, using columnar file format, and compression in conjunction with bucketing. This can enable your data to further enhance query performance.
Happy bucketing!
About the Authors Takeshi Nakatani is a Principal Big Data Consultant on the Professional Services team in Tokyo. He has 26 years of experience in the IT industry, with expertise in architecting data infrastructure. On his days off, he can be a rock drummer or a motorcyclist. Noritaka Sekiyama is a Principal Big Data Architect on the AWS Glue team. He is responsible for building software artifacts to help customers. In his spare time, he enjoys cycling with his road bike.