Post Syndicated from Takeshi Nakatani original https://aws.amazon.com/blogs/big-data/optimize-data-layout-by-bucketing-with-amazon-athena-and-aws-glue-to-accelerate-downstream-queries/
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_name
WHERE
report_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:
- Open the Athena console.
- In the query editor, run the following DDL to create a new AWS Glue database:
-- Create Glue database
CREATE DATABASE bucketing_blog;
- For Database under Data, choose
bucketing_blog to set the current database.
- Run the following DDL to create the original table:
-- Create original table
CREATE 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.
- Run the following DDL to add partitions. We add partitions only for 5 years out of 124 years based on the use case requirement:
-- Load partitions
ALTER 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/';
- Run the following DML to verify if you can successfully query the data:
-- 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.
- Run a query against the original table to evaluate the query performance as a baseline. The following query selects records for five specific stations with report type
CRN05:
-- Baseline
SELECT station, report_type, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, tmp
FROM "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.
- Open the Athena query editor.
- Run the following query, providing your own S3 bucket and prefix:
--CTAS, non-bucketed
CREATE 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'
)
AS
SELECT
station, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, vis, tmp, dew, slp, aj1, gf1, mw1, report_type
FROM "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:
-- Baseline
SELECT station, report_type, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, tmp
FROM "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.
- Run the following query:
-- CTAS, Hive-bucketed
CREATE 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'
)
AS
SELECT
station, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, vis, tmp, dew, slp, aj1, gf1, mw1, report_type
FROM "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.
- Query the non-bucketed table with the following statement:
-- No bucketing
SELECT station, report_type, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, tmp
FROM "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.
- Query the bucketed table with the following statement:
-- Hive bucketing
SELECT station, report_type, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, tmp
FROM "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:
- On the AWS Glue console, choose ETL jobs in the navigation pane.
- Choose Create job and choose Visual ETL.
- Under Add nodes, choose AWS Glue Data Catalog for Sources.
- For Database, choose
bucketing_blog.
- For Table, choose
noaa_remote_original.
- Under Add nodes, choose Change Schema for Transforms.
- Under Add nodes, choose Custom Transform for Transforms.
- For Name, enter
ToS3WithBucketing.
- For Node parents, choose Change Schema.
- For Code block, enter the following code snippet:
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:
- The AWS Glue Data Catalog node loads the
noaa_remote_original table from the Data Catalog
- The Change Schema node makes sure that it loads columns registered in the Data Catalog
- The ToS3WithBucketing node writes data to Amazon S3 with both partitioning and Spark-based bucketing
The job has been successfully authored in the visual editor.
- Under Job details, for IAM Role, choose your AWS Identity and Access Management (IAM) role for this job.
- For Worker type, choose G.8X.
- For Requested number of workers, enter 5.
- Choose Save, then choose Run.
After these steps, the table glue_bucketed. has been created.
- Choose Tables in the navigation pane, and choose the table
glue_bucketed.
- On the Actions menu, choose Edit table under Manage.
- In the Table properties section, choose Add.
- Add a key pair with key
bucketing_format and value spark.
- Choose Save.
Now it’s time to query the tables.
- Query the bucketed table with the following statement:
-- Spark bucketing
SELECT station, report_type, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, tmp
FROM "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:
- Open the Athena query editor.
- Run the following query to create an Iceberg table with hidden partitioning along with bucketing:
-- CTAS, Iceberg-bucketed
CREATE 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'
)
AS
SELECT
station, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, vis, tmp, dew, slp, aj1, gf1, mw1, report_type
FROM "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.
- Query the bucketed table with the following statement:
-- Iceberg bucketing
SELECT station, report_type, date, source, latitude, longitude, elevation, name, call_sign, quality_control, wnd, cig, tmp
FROM "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.
Sidhanth Muralidhar is a Principal Technical Account Manager at AWS. He works with large enterprise customers who run their workloads on AWS. He is passionate about working with customers and helping them architect workloads for costs, reliability, performance, and operational excellence at scale in their cloud journey. He has a keen interest in data analytics as well.
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.
