AWS-powered knowledge lakes, supported by the unrivaled availability of Amazon Easy Storage Service (Amazon S3), can deal with the dimensions, agility, and adaptability required to mix totally different knowledge and analytics approaches. As knowledge lakes have grown in measurement and matured in utilization, a big quantity of effort could be spent preserving the information in line with enterprise occasions. To make sure information are up to date in a transactionally constant method, a rising variety of clients are utilizing open-source transactional desk codecs akin to Apache Iceberg, Apache Hudi, and Linux Basis Delta Lake that provide help to retailer knowledge with excessive compression charges, natively interface along with your purposes and frameworks, and simplify incremental knowledge processing in knowledge lakes constructed on Amazon S3. These codecs allow ACID (atomicity, consistency, isolation, sturdiness) transactions, upserts, and deletes, and superior options akin to time journey and snapshots that have been beforehand solely obtainable in knowledge warehouses. Every storage format implements this performance in barely alternative ways; for a comparability, discuss with Selecting an open desk format in your transactional knowledge lake on AWS.
In 2023, AWS introduced basic availability for Apache Iceberg, Apache Hudi, and Linux Basis Delta Lake in Amazon Athena for Apache Spark, which removes the necessity to set up a separate connector or related dependencies and handle variations, and simplifies the configuration steps required to make use of these frameworks.
On this submit, we present you how one can use Spark SQL in Amazon Athena notebooks and work with Iceberg, Hudi, and Delta Lake desk codecs. We display frequent operations akin to creating databases and tables, inserting knowledge into the tables, querying knowledge, and snapshots of the tables in Amazon S3 utilizing Spark SQL in Athena.
Stipulations
Full the next conditions:
Obtain and import instance notebooks from Amazon S3
To observe alongside, obtain the notebooks mentioned on this submit from the next places:
After you obtain the notebooks, import them into your Athena Spark atmosphere by following the To import a pocket book part in Managing pocket book information.
Navigate to particular Open Desk Format part
If you’re concerned about Iceberg desk format, navigate to Working with Apache Iceberg tables part.
If you’re concerned about Hudi desk format, navigate to Working with Apache Hudi tables part.
If you’re concerned about Delta Lake desk format, navigate to Working with Linux basis Delta Lake tables part.
Working with Apache Iceberg tables
When utilizing Spark notebooks in Athena, you possibly can run SQL queries immediately with out having to make use of PySpark. We do that by utilizing cell magics, that are particular headers in a pocket book cell that change the cell’s habits. For SQL, we are able to add the %%sql
magic, which is able to interpret your entire cell contents as a SQL assertion to be run on Athena.
On this part, we present how you need to use SQL on Apache Spark for Athena to create, analyze, and handle Apache Iceberg tables.
Arrange a pocket book session
So as to use Apache Iceberg in Athena, whereas creating or modifying a session, choose the Apache Iceberg choice by increasing the Apache Spark properties part. It should pre-populate the properties as proven within the following screenshot.
For steps, see Modifying session particulars or Creating your individual pocket book.
The code used on this part is accessible within the SparkSQL_iceberg.ipynb file to observe alongside.
Create a database and Iceberg desk
First, we create a database within the AWS Glue Information Catalog. With the next SQL, we are able to create a database referred to as icebergdb
:
Subsequent, within the database icebergdb
, we create an Iceberg desk referred to as noaa_iceberg
pointing to a location in Amazon S3 the place we’ll load the information. Run the next assertion and exchange the situation s3://<your-S3-bucket>/<prefix>/
along with your S3 bucket and prefix:
Insert knowledge into the desk
To populate the noaa_iceberg
Iceberg desk, we insert knowledge from the Parquet desk sparkblogdb.noaa_pq
that was created as a part of the conditions. You are able to do this utilizing an INSERT INTO assertion in Spark:
Alternatively, you need to use CREATE TABLE AS SELECT with the USING iceberg clause to create an Iceberg desk and insert knowledge from a supply desk in a single step:
Question the Iceberg desk
Now that the information is inserted within the Iceberg desk, we are able to begin analyzing it. Let’s run a Spark SQL to seek out the minimal recorded temperature by 12 months for the 'SEATTLE TACOMA AIRPORT, WA US'
location:
We get following output.
Replace knowledge within the Iceberg desk
Let’s take a look at how one can replace knowledge in our desk. We wish to replace the station identify 'SEATTLE TACOMA AIRPORT, WA US'
to 'Sea-Tac'
. Utilizing Spark SQL, we are able to run an UPDATE assertion in opposition to the Iceberg desk:
We are able to then run the earlier SELECT question to seek out the minimal recorded temperature for the 'Sea-Tac'
location:
We get the next output.
Compact knowledge information
Open desk codecs like Iceberg work by creating delta adjustments in file storage, and monitoring the variations of rows by manifest information. Extra knowledge information results in extra metadata saved in manifest information, and small knowledge information typically trigger an pointless quantity of metadata, leading to much less environment friendly queries and better Amazon S3 entry prices. Operating Iceberg’s rewrite_data_files
process in Spark for Athena will compact knowledge information, combining many small delta change information right into a smaller set of read-optimized Parquet information. Compacting information hurries up the learn operation when queried. To run compaction on our desk, run the next Spark SQL:
rewrite_data_files presents choices to specify your kind technique, which may help reorganize and compact knowledge.
Record desk snapshots
Every write, replace, delete, upsert, and compaction operation on an Iceberg desk creates a brand new snapshot of a desk whereas preserving the previous knowledge and metadata round for snapshot isolation and time journey. To record the snapshots of an Iceberg desk, run the next Spark SQL assertion:
Expire previous snapshots
Commonly expiring snapshots is really helpful to delete knowledge information which are not wanted, and to maintain the dimensions of desk metadata small. It should by no means take away information which are nonetheless required by a non-expired snapshot. In Spark for Athena, run the next SQL to run out snapshots for the desk icebergdb.noaa_iceberg
which are older than a particular timestamp:
Observe that the timestamp worth is specified as a string in format yyyy-MM-dd HH:mm:ss.fff
. The output will give a depend of the variety of knowledge and metadata information deleted.
Drop the desk and database
You’ll be able to run the next Spark SQL to scrub up the Iceberg tables and related knowledge in Amazon S3 from this train:
Run the next Spark SQL to take away the database icebergdb:
To study extra about all of the operations you possibly can carry out on Iceberg tables utilizing Spark for Athena, discuss with Spark Queries and Spark Procedures within the Iceberg documentation.
Working with Apache Hudi tables
Subsequent, we present how you need to use SQL on Spark for Athena to create, analyze, and handle Apache Hudi tables.
Arrange a pocket book session
So as to use Apache Hudi in Athena, whereas creating or modifying a session, choose the Apache Hudi choice by increasing the Apache Spark properties part.
For steps, see Modifying session particulars or Creating your individual pocket book.
The code used on this part needs to be obtainable within the SparkSQL_hudi.ipynb file to observe alongside.
Create a database and Hudi desk
First, we create a database referred to as hudidb
that can be saved within the AWS Glue Information Catalog adopted by Hudi desk creation:
We create a Hudi desk pointing to a location in Amazon S3 the place we’ll load the information. Observe that the desk is of copy-on-write kind. It’s outlined by kind="cow"
within the desk DDL. Now we have outlined station and date because the a number of main keys and preCombinedField as 12 months. Additionally, the desk is partitioned on 12 months. Run the next assertion and exchange the situation s3://<your-S3-bucket>/<prefix>/
along with your S3 bucket and prefix:
Insert knowledge into the desk
Like with Iceberg, we use the INSERT INTO assertion to populate the desk by studying knowledge from the sparkblogdb.noaa_pq
desk created within the earlier submit:
Question the Hudi desk
Now that the desk is created, let’s run a question to seek out the utmost recorded temperature for the 'SEATTLE TACOMA AIRPORT, WA US'
location:
Replace knowledge within the Hudi desk
Let’s change the station identify 'SEATTLE TACOMA AIRPORT, WA US'
to 'Sea–Tac'
. We are able to run an UPDATE assertion on Spark for Athena to replace the information of the noaa_hudi
desk:
We run the earlier SELECT question to seek out the utmost recorded temperature for the 'Sea-Tac'
location:
Run time journey queries
We are able to use time journey queries in SQL on Athena to research previous knowledge snapshots. For instance:
This question checks the Seattle Airport temperature knowledge as of a particular time previously. The timestamp clause lets us journey again with out altering present knowledge. Observe that the timestamp worth is specified as a string in format yyyy-MM-dd HH:mm:ss.fff
.
Optimize question pace with clustering
To enhance question efficiency, you possibly can carry out clustering on Hudi tables utilizing SQL in Spark for Athena:
Compact tables
Compaction is a desk service employed by Hudi particularly in Merge On Learn (MOR) tables to merge updates from row-based log information to the corresponding columnar-based base file periodically to provide a brand new model of the bottom file. Compaction shouldn’t be relevant to Copy On Write (COW) tables and solely applies to MOR tables. You’ll be able to run the next question in Spark for Athena to carry out compaction on MOR tables:
Drop the desk and database
Run the next Spark SQL to take away the Hudi desk you created and related knowledge from the Amazon S3 location:
Run the next Spark SQL to take away the database hudidb
:
To find out about all of the operations you possibly can carry out on Hudi tables utilizing Spark for Athena, discuss with SQL DDL and Procedures within the Hudi documentation.
Working with Linux basis Delta Lake tables
Subsequent, we present how you need to use SQL on Spark for Athena to create, analyze, and handle Delta Lake tables.
Arrange a pocket book session
So as to use Delta Lake in Spark for Athena, whereas creating or modifying a session, choose Linux Basis Delta Lake by increasing the Apache Spark properties part.
For steps, see Modifying session particulars or Creating your individual pocket book.
The code used on this part needs to be obtainable within the SparkSQL_delta.ipynb file to observe alongside.
Create a database and Delta Lake desk
On this part, we create a database within the AWS Glue Information Catalog. Utilizing following SQL, we are able to create a database referred to as deltalakedb
:
Subsequent, within the database deltalakedb
, we create a Delta Lake desk referred to as noaa_delta
pointing to a location in Amazon S3 the place we’ll load the information. Run the next assertion and exchange the situation s3://<your-S3-bucket>/<prefix>/
along with your S3 bucket and prefix:
Insert knowledge into the desk
We use an INSERT INTO assertion to populate the desk by studying knowledge from the sparkblogdb.noaa_pq
desk created within the earlier submit:
You can even use CREATE TABLE AS SELECT to create a Delta Lake desk and insert knowledge from a supply desk in a single question.
Question the Delta Lake desk
Now that the information is inserted within the Delta Lake desk, we are able to begin analyzing it. Let’s run a Spark SQL to seek out the minimal recorded temperature for the 'SEATTLE TACOMA AIRPORT, WA US'
location:
Replace knowledge within the Delta lake desk
Let’s change the station identify 'SEATTLE TACOMA AIRPORT, WA US'
to 'Sea–Tac'
. We are able to run an UPDATE assertion on Spark for Athena to replace the information of the noaa_delta
desk:
We are able to run the earlier SELECT question to seek out the minimal recorded temperature for the 'Sea-Tac'
location, and the consequence needs to be the identical as earlier:
Compact knowledge information
In Spark for Athena, you possibly can run OPTIMIZE on the Delta Lake desk, which is able to compact the small information into bigger information, so the queries aren’t burdened by the small file overhead. To carry out the compaction operation, run the next question:
Check with Optimizations within the Delta Lake documentation for various choices obtainable whereas working OPTIMIZE.
Take away information not referenced by a Delta Lake desk
You’ll be able to take away information saved in Amazon S3 which are not referenced by a Delta Lake desk and are older than the retention threshold by working the VACCUM command on the desk utilizing Spark for Athena:
Check with Take away information not referenced by a Delta desk within the Delta Lake documentation for choices obtainable with VACUUM.
Drop the desk and database
Run the next Spark SQL to take away the Delta Lake desk you created:
Run the next Spark SQL to take away the database deltalakedb
:
Operating DROP TABLE DDL on the Delta Lake desk and database deletes the metadata for these objects, however doesn’t mechanically delete the information information in Amazon S3. You’ll be able to run the next Python code within the pocket book’s cell to delete the information from the S3 location:
To study extra in regards to the SQL statements you can run on a Delta Lake desk utilizing Spark for Athena, discuss with the quickstart within the Delta Lake documentation.
Conclusion
This submit demonstrated how one can use Spark SQL in Athena notebooks to create databases and tables, insert and question knowledge, and carry out frequent operations like updates, compactions, and time journey on Hudi, Delta Lake, and Iceberg tables. Open desk codecs add ACID transactions, upserts, and deletes to knowledge lakes, overcoming limitations of uncooked object storage. By eradicating the necessity to set up separate connectors, Spark on Athena’s built-in integration reduces configuration steps and administration overhead when utilizing these in style frameworks for constructing dependable knowledge lakes on Amazon S3. To study extra about deciding on an open desk format in your knowledge lake workloads, discuss with Selecting an open desk format in your transactional knowledge lake on AWS.
In regards to the Authors
Pathik Shah is a Sr. Analytics Architect on Amazon Athena. He joined AWS in 2015 and has been focusing within the large knowledge analytics area since then, serving to clients construct scalable and strong options utilizing AWS analytics companies.
Raj Devnath is a Product Supervisor at AWS on Amazon Athena. He’s enthusiastic about constructing merchandise clients love and serving to clients extract worth from their knowledge. His background is in delivering options for a number of finish markets, akin to finance, retail, sensible buildings, dwelling automation, and knowledge communication techniques.