Construct a transactional knowledge lake utilizing Apache Iceberg, AWS Glue, and cross-account knowledge shares utilizing AWS Lake Formation and Amazon Athena

Constructing a knowledge lake on Amazon Easy Storage Service (Amazon S3) offers quite a few advantages for a corporation. It lets you entry various knowledge sources, construct enterprise intelligence dashboards, construct AI and machine studying (ML) fashions to offer custom-made buyer experiences, and speed up the curation of latest datasets for consumption by adopting a fashionable knowledge structure or knowledge mesh structure.

Nevertheless, many use circumstances, like performing change knowledge seize (CDC) from an upstream relational database to an Amazon S3-based knowledge lake, require dealing with knowledge at a report degree. Performing an operation like inserting, updating, and deleting particular person data from a dataset requires the processing engine to learn all of the objects (information), make the modifications, and rewrite whole datasets as new information. Moreover, making the information accessible within the knowledge lake in near-real time typically results in the information being fragmented over many small information, leading to poor question efficiency and compaction upkeep.

In 2022, we introduced which you could implement fine-grained entry management insurance policies utilizing AWS Lake Formation and question knowledge saved in any supported file format utilizing desk codecs equivalent to Apache Iceberg, Apache Hudi, and extra utilizing Amazon Athena queries. You get the pliability to decide on the desk and file format finest suited in your use case and get the advantage of centralized knowledge governance to safe knowledge entry when utilizing Athena.

On this put up, we present you configure Lake Formation utilizing Iceberg desk codecs. We additionally clarify upsert and merge in an S3 knowledge lake utilizing an Iceberg framework and apply Lake Formation entry management utilizing Athena.

Iceberg is an open desk format for very massive analytic datasets. Iceberg manages massive collections of information as tables, and it helps fashionable analytical knowledge lake operations equivalent to record-level insert, replace, delete, and time journey queries. The Iceberg specification permits seamless desk evolution equivalent to schema and partition evolution, and its design is optimized for utilization on Amazon S3. Iceberg additionally helps assure knowledge correctness below concurrent write eventualities.

Answer overview

To clarify this setup, we current the next structure, which integrates Amazon S3 for the information lake (Iceberg desk format), Lake Formation for entry management, AWS Glue for ETL (extract, rework, and cargo), and Athena for querying the most recent stock knowledge from the Iceberg tables utilizing commonplace SQL.

The answer workflow consists of the next steps, together with knowledge ingestion (Steps 1–3), knowledge governance (Step 4), and knowledge entry (Step 5):

1. We use AWS Database Migration Service (AWS DMS) or the same instrument to hook up with the information supply and transfer incremental knowledge (CDC) to Amazon S3 in CSV format.
2. An AWS Glue PySpark job reads the incremental knowledge from the S3 enter bucket and performs deduplication of the data.
3. The job then invokes Iceberg’s MERGE statements to merge the information with the goal S3 bucket.
4. We use the AWS Glue Information Catalog as a centralized catalog, which is utilized by AWS Glue and Athena. An AWS Glue crawler is built-in on high of S3 buckets to routinely detect the schema. Lake Formation lets you centrally handle permissions and entry management for Information Catalog assets in your S3 knowledge lake. You should utilize fine-grained entry management in Lake Formation to limit entry to knowledge in question outcomes.
5. We use Athena built-in with Lake Formation to question knowledge from the Iceberg desk utilizing commonplace SQL and validate table- and column-level entry on Iceberg tables.

For this answer, we assume that the uncooked knowledge information are already accessible in Amazon S3, and concentrate on processing the information utilizing AWS Glue with Iceberg desk format. We use pattern merchandise knowledge that has the next attributes:

• op – This represents the operation on the supply report. This reveals values I to signify insert operations, U to signify updates, and D to signify deletes. You’ll want to be certain this attribute is included in your CDC incremental knowledge earlier than it will get written to Amazon S3. Ensure you seize this attribute, in order that your ETL logic can take acceptable motion whereas merging it.
• product_id – That is the first key column within the supply knowledge desk.
• class – This column represents the class of an merchandise.
• product_name – That is the title of the product.
• quantity_available – That is the amount accessible within the stock. After we showcase the incremental knowledge for UPSERT or MERGE, we scale back the amount accessible for the product to showcase the performance.
• last_update_time – That is the time when the merchandise report was up to date on the supply knowledge.

We display implementing the answer with the next steps:

1. Create an S3 bucket for enter and output knowledge.
2. Create enter and output tables utilizing Athena.
3. Insert the information into the Iceberg desk from Athena.
4. Question the Iceberg desk utilizing Athena.
5. Add incremental (CDC) knowledge for additional processing.
6. Run the AWS Glue job once more to course of the incremental information.
7. Question the Iceberg desk once more utilizing Athena.
8. Outline Lake Formation insurance policies.

Stipulations

For Athena queries, we have to configure an Athena workgroup with engine model 3 to assist Iceberg desk format.

To validate cross-account entry by way of Lake Formation for Iceberg desk, on this put up we used two accounts (main and secondary).

Now let’s dive into the implementation steps.

Create an S3 bucket for enter and output knowledge

Earlier than we run the AWS Glue job, we’ve got to add the pattern CSV information to the enter bucket and course of them with AWS Glue PySpark code for the output.

To create an S3 bucket, full the next steps:

1. On the Amazon S3 console, select Buckets within the navigation pane.
2. Select Create bucket.
3. Specify the bucket title asiceberg-blog and go away the remaining fields as default.

S3 bucket names are globally distinctive. Whereas implementing the answer, you could get an error saying the bucket title already exists. Make certain to offer a singular title and use the identical title whereas implementing the remainder of the implementation steps. Formatting the bucket title as<Bucket-Identify>-${AWS_ACCOUNT_ID}-${AWS_REGION_CODE}would possibly show you how to get a singular title.

4. On the bucket particulars web page, select Create folder.
5. Create two subfolders. For this put up, we createiceberg-blog/raw-csv-input andiceberg-blog/iceberg-output.
   
6. Add theLOAD00000001.csvfile into the raw-csv-input folder.

The next screenshot offers a pattern of the enter dataset.

Create enter and output tables utilizing Athena

To create enter and output Iceberg tables within the AWS Glue Information Catalog, open the Athena question editor and run the next queries in sequence:

-- Create database for the demo
CREATE DATABASE iceberg_lf_db;

As we clarify later on this put up, it’s important to report the information areas when incorporating Lake Formation entry controls.

-- Create exterior desk in enter CSV information. Exchange the S3 path along with your bucket title
CREATE EXTERNAL TABLE iceberg_lf_db.csv_input(
op string,
product_id bigint,
class string,
product_name string,
quantity_available bigint,
last_update_time string)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
STORED AS INPUTFORMAT 'org.apache.hadoop.mapred.TextInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION 's3://glue-iceberg-demo/raw-csv-input/'
TBLPROPERTIES (
'areColumnsQuoted'='false',
'classification'='csv',
'columnsOrdered'='true',
'compressionType'='none',
'delimiter'=',',
'typeOfData'='file');

-- Create output Iceberg desk with partitioning. Exchange the S3 bucket title along with your bucket title
CREATE TABLE iceberg_lf_db.iceberg_table_lf (
product_id bigint,
class string,
product_name string,
quantity_available bigint,
last_update_time timestamp)
PARTITIONED BY (class, bucket(16,product_id))
LOCATION 's3://glue-iceberg-demo/iceberg_blog/iceberg-output/'
TBLPROPERTIES (
'table_type'='ICEBERG',
'format'='parquet',
'write_target_data_file_size_bytes'='536870912'
);

-- Validate the enter knowledge
SELECT * FROM iceberg_lf_db.csv_input;

SELECT * FROM iceberg_lf_db.iceberg_table_lf;

Alternatively, you need to use an AWS Glue crawler to create the desk definition for the enter information.

Insert the information into the Iceberg desk from Athena

Optionally, we will insert knowledge into the Iceberg desk by way of Athena utilizing the next code:

insert into iceberg_lf_demo.iceberg_lf_output_athena (product_id,class,product_name,quantity_available,last_update_time) values (200,'Cell','Cell model 1',25,forged('2023-01-19 09:51:40' as timestamp));
insert into iceberg_lf_demo.iceberg_lf_output_athena (product_id,class,product_name,quantity_available,last_update_time) values (201,'Laptop computer','Laptop computer model 1',20,forged('2023-01-19 09:51:40' as timestamp));
insert into iceberg_lf_demo.iceberg_lf_output_athena (product_id,class,product_name,quantity_available,last_update_time) values (202,'Pill','Kindle',30,forged('2023-01-19 09:51:41' as timestamp));
insert into iceberg_lf_demo.iceberg_lf_output_athena (product_id,class,product_name,quantity_available,last_update_time) values (203,'Speaker','Alexa',10,forged('2023-01-19 09:51:42' as timestamp));
insert into iceberg_lf_demo.iceberg_lf_output_athena (product_id,class,product_name,quantity_available,last_update_time) values (204,'Speaker','Alexa',50,forged('2023-01-19 09:51:43' as timestamp));

For this put up, we load the information utilizing an AWS Glue job. Full the next steps to create the job:

1. On the AWS Glue console, select Jobs within the navigation pane.
2. Select Create job.
3. Choose Visible with a clean canvas.
4. Select Create.
   
5. Select Edit script.
   
6. Exchange the script with the next script:

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

from pyspark.sql.capabilities import *
from awsglue.dynamicframe import DynamicFrame

from pyspark.sql.window import Window
from pyspark.sql.capabilities import rank, max

from pyspark.conf import SparkConf

args = getResolvedOptions(sys.argv, ["JOB_NAME"])
conf = SparkConf()

## spark.sql.catalog.job_catalog.warehouse may be handed as an ## runtime argument with worth because the S3 path
## Please be certain to move runtime argument –
## iceberg_job_catalog_warehouse with worth because the S3 path 
conf.set("spark.sql.catalog.job_catalog.warehouse", args['iceberg_job_catalog_warehouse'])
conf.set("spark.sql.catalog.job_catalog", "org.apache.iceberg.spark.SparkCatalog")
conf.set("spark.sql.catalog.job_catalog.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog")
conf.set("spark.sql.catalog.job_catalog.io-impl", "org.apache.iceberg.aws.s3.S3FileIO")
conf.set("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
conf.set("spark.sql.iceberg.handle-timestamp-without-timezone","true")

sc = SparkContext(conf=conf)
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args["JOB_NAME"], args)


## Learn Enter Desk
## glueContext.create_data_frame.from_catalog may be extra 
## performant and may be changed rather than 
## create_dynamic_frame.from_catalog.

IncrementalInputDyF = glueContext.create_dynamic_frame.from_catalog(database = "iceberg_lf_db", table_name = "csv_input", transformation_ctx = "IncrementalInputDyF")
IncrementalInputDF = IncrementalInputDyF.toDF()

if not IncrementalInputDF.rdd.isEmpty():
## Apply De-duplication logic on enter knowledge, to pickup newest report primarily based on timestamp and operation
IDWindowDF = Window.partitionBy(IncrementalInputDF.product_id).orderBy(IncrementalInputDF.last_update_time).rangeBetween(-sys.maxsize, sys.maxsize)

# Add new columns to seize OP worth and what's the newest timestamp
inputDFWithTS= IncrementalInputDF.withColumn("max_op_date",max(IncrementalInputDF.last_update_time).over(IDWindowDF))

# Filter out new data which are inserted, then choose newest report from present data and merge each to get deduplicated output
NewInsertsDF = inputDFWithTS.filter("last_update_time=max_op_date").filter("op='I'")
UpdateDeleteDf = inputDFWithTS.filter("last_update_time=max_op_date").filter("op IN ('U','D')")
finalInputDF = NewInsertsDF.unionAll(UpdateDeleteDf)

# Register the deduplicated enter as short-term desk to make use of in Iceberg Spark SQL statements
finalInputDF.createOrReplaceTempView("incremental_input_data")
finalInputDF.present()

## Carry out merge operation on incremental enter knowledge with MERGE INTO. This part of the code makes use of Spark SQL to showcase the expressive SQL strategy of Iceberg to carry out a Merge operation
IcebergMergeOutputDF = spark.sql("""
MERGE INTO job_catalog.iceberg_lf_db.iceberg_table_lf t
USING (SELECT op, product_id, class, product_name, quantity_available, to_timestamp(last_update_time) as last_update_time FROM incremental_input_data) s
ON t.product_id = s.product_id
WHEN MATCHED AND s.op = 'D' THEN DELETE
WHEN MATCHED THEN UPDATE SET t.quantity_available = s.quantity_available, t.last_update_time = s.last_update_time
WHEN NOT MATCHED THEN INSERT (product_id, class, product_name, quantity_available, last_update_time) VALUES (s.product_id, s.class, s.product_name, s.quantity_available, s.last_update_time)
""")

job.commit()

7. On the Job particulars tab, specify the job title (iceberg-lf).
8. For IAM Function, assign an AWS Id and Entry Administration (IAM) position that has the required permissions to run an AWS Glue job and browse and write to the S3 bucket.
9. For Glue model, select Glue 4.0 (Glue 3.0 can also be supported).
10. For Language, select Python 3.
11. Make certain Job bookmark has the default worth of Allow.
    
12. For Job parameters, add the next:
    1. Add the important thing--datalake-formatswith the worthiceberg.
    2. Add the important thing--iceberg_job_catalog_warehouse with the worth as your S3 path (s3://<bucket-name>/<iceberg-warehouse-path>).
       
13. Select Save after which Run, which ought to write the enter knowledge to the Iceberg desk with a MERGE assertion.

Question the Iceberg desk utilizing Athena

After you have got efficiently run the AWS Glue job, you’ll be able to validate the output in Athena with the next SQL question:

SELECT * FROM iceberg_lf_db.iceberg_table_lf restrict 10;

The output of the question ought to match the enter, with one distinction: the Iceberg output desk doesn’t have theopcolumn.

Add incremental (CDC) knowledge for additional processing

After we course of the preliminary full load file, let’s add an incremental file.

This file consists of up to date data on two objects.

Run the AWS Glue job once more to course of incremental information

As a result of the AWS Glue job has bookmarks enabled, the job picks up the brand new incremental file and performs a MERGE operation on the Iceberg desk.

To run the job once more, full the next steps:

1. On the AWS Glue console, select Jobs within the navigation pane.
2. Choose the job and select Run.

For this put up, we run the job manually, however you’ll be able to configure your AWS Glue jobs to run as a part of an AWS Glue workflow or by way of AWS Step Capabilities (for extra data, see Handle AWS Glue Jobs with Step Capabilities).

Question the Iceberg desk utilizing Athena after incremental knowledge processing

When the incremental knowledge processing is full, you’ll be able to run the identical SELECT assertion once more and validate that the amount worth is up to date for objects 200 and 201.

The next screenshot reveals the output.

Outline Lake Formation insurance policies

For knowledge governance, we use Lake Formation. Lake Formation is a totally managed service that simplifies knowledge lake setup, helps centralized safety administration, and offers transactional entry on high of your knowledge lake. Furthermore, it allows knowledge sharing throughout accounts and organizations. There are two methods to share knowledge assets in Lake Formation: named useful resource entry management (NRAC) and tag-based entry management (TBAC). NRAC makes use of AWS Useful resource Entry Supervisor (AWS RAM) to share knowledge assets throughout accounts utilizing Lake Formation V3. These are consumed by way of useful resource hyperlinks which are primarily based on created useful resource shares. Lake Formation tag-based entry management (LF-TBAC) is one other strategy to share knowledge assets in Lake Formation, which defines permissions primarily based on attributes. These attributes are referred to as LF-tags.

On this instance, we create databases within the main account. Our NRAC database is shared with a knowledge area by way of AWS RAM. Entry to knowledge tables that we register on this database shall be dealt with by way of NRAC.

Configure entry controls within the main account

Within the main account, full the next steps to arrange entry controls utilizing Lake Formation:

1. On the Lake Formation console, select Information lake areas within the navigation pane.
2. Select Register location.
3. Replace the Iceberg Amazon S3 location path proven within the following screenshot.
   

Grant entry to the database to the secondary account

To grant database entry to the exterior (secondary) account, full the next steps:

1. On the Lake Formation console, navigate to your database.
2. On the Actions menu, select Grant.
   
3. Select Exterior accounts and enter the secondary account quantity.
4. Choose Named knowledge catalog assets.
5. Confirm the database title.

The primary grant needs to be at database degree, and the second grant is at desk degree.

6. For Database permissions, specify your permissions (for this put up, we choose Describe).
7. Select Grant.
   

Now you could grant permissions on the desk degree.

8. Choose Exterior accounts and enter the secondary account quantity.
9. Choose Named knowledge catalog assets.
10. Confirm the desk title.
11. For Desk permissions, specify the permissions you wish to grant. For this put up, we choose Choose and Describe.
12. Select Grant.
    

If you happen to see the next error, you should revokeIAMAllowedPrincipalsfrom the information lake permissions.

To take action, choose IAMAllowedPrincipals and select Revoke.

Select Revoke once more to substantiate.

After you revoke the information permissions, the permissions ought to seem as proven within the following screenshot.

Add AWS Glue IAM position permissions

As a result of the IAM principal position was revoked, the AWS Glue IAM position that was used within the AWS Glue job must be added completely to grant entry as proven within the following screenshot.

You’ll want to repeat these steps for the AWS Glue IAM position at desk degree.

Confirm the permissions granted to the AWS Glue IAM position on the Lake Formation console.

Grant entry to the Iceberg desk to the exterior account

Within the secondary account, full the next steps to grant entry to the Iceberg desk to exterior account.

1. On the AWS RAM console, select Useful resource shares within the navigation pane.
2. Select the useful resource shares invitation despatched from the first account.
   
3. Select Settle for useful resource share.
   

The useful resource standing ought to now be energetic.

Subsequent, you could create a useful resource hyperlink for the shared Iceberg desk and entry by way of Athena.

4. On the Lake Formation console, select Tables within the navigation pane.
5. Choose the Iceberg desk (shared from the first account).
6. On the Actions menu, select Create useful resource hyperlink.
   
7. For Useful resource hyperlink title, enter a reputation (for this put up,iceberg_table_lf_demo).
8. For Database, select your database and confirm the shared desk and database are routinely populated.
9. Select Create.
   
10. Choose your desk and on the Actions menu, select View knowledge.
    

You’re redirected to the Athena console, the place you’ll be able to question the information.

Grant column-based entry within the main account

For column-level restricted entry, you could grant entry on the column degree on the Iceberg desk. Full the next steps:

1. On the Lake Formation console, navigate to your database.
2. On the Actions menu, select Grant.
3. Choose Exterior accounts and enter the secondary account quantity.
4. Choose Named knowledge catalog assets.
5. Confirm the desk title.
6. For Desk permissions, select the permissions you wish to grant. For this put up, we choose Choose.
7. Below Information permissions, select Column-based entry.
8. Choose Embrace columns and select your permission filters (for this put up, Class and Quantity_available).
9. Select Grant.
   

Information with restricted columns can now be queried by way of the Athena console.

Clear up

To keep away from incurring ongoing prices, full the next steps to wash up your assets:

1. In your secondary account, log in to the Lake Formation console.
2. Drop the useful resource share desk.
3. In your main account, log in to the Lake Formation console.
4. Revoke the entry you configured.
5. Drop the AWS Glue tables and database.
6. Delete the AWS Glue job.
7. Delete the S3 buckets and every other assets that you just created as a part of the stipulations for this put up.

Conclusion

This put up explains how you need to use the Iceberg framework with AWS Glue and Lake Formation to outline cross-account entry controls and question knowledge utilizing Athena. It offers an outline of Iceberg and its options and integration approaches, and explains how one can ingest knowledge, grant cross-account entry, and question knowledge by way of a step-by-step information.

We hope this offers you an excellent start line for utilizing Iceberg to construct your knowledge lake platform together with AWS analytics providers to implement your answer.

Concerning the Authors

Vikram Sahadevan is a Senior Resident Architect on the AWS Information Lab crew. He enjoys efforts that focus round offering prescriptive architectural steerage, sharing finest practices, and eradicating technical roadblocks with joint engineering engagements between prospects and AWS technical assets that speed up knowledge, analytics, synthetic intelligence, and machine studying initiatives.

Suvendu Kumar Patra possesses 18 years of expertise in infrastructure, database design, and knowledge engineering, and he presently holds the place of Senior Resident Architect at Amazon Internet Companies. He’s a member of the specialised focus group, AWS Information Lab, and his main duties entail working with govt management groups of strategic AWS prospects to develop their roadmaps for knowledge, analytics, and AI/ML. Suvendu collaborates carefully with prospects to implement knowledge engineering, knowledge hub, knowledge lake, knowledge governance, and EDW options, in addition to enterprise knowledge technique and knowledge administration.