Speed up light-weight analytics utilizing PyIceberg with AWS Lambda and an AWS Glue Iceberg REST endpoint

For contemporary organizations constructed on information insights, efficient information administration is essential for powering superior analytics and machine studying (ML) actions. As information use circumstances change into extra advanced, information engineering groups require subtle tooling to deal with versioning, rising information volumes, and schema adjustments throughout a number of information sources and purposes.

Apache Iceberg has emerged as a preferred selection for information lakes, providing ACID (Atomicity, Consistency, Isolation, Sturdiness) transactions, schema evolution, and time journey capabilities. Iceberg tables might be accessed from numerous distributed information processing frameworks like Apache Spark and Trino, making it a versatile answer for numerous information processing wants. Among the many out there instruments for working with Iceberg, PyIceberg stands out as a Python implementation that permits desk entry and administration with out requiring distributed compute assets.

On this submit, we exhibit how PyIceberg, built-in with the AWS Glue Knowledge Catalog and AWS Lambda, supplies a light-weight method to harness Iceberg’s highly effective options by intuitive Python interfaces. We present how this integration permits groups to start out working with Iceberg tables with minimal setup and infrastructure dependencies.

PyIceberg’s key capabilities and benefits

One in every of PyIceberg’s main benefits is its light-weight nature. With out requiring distributed computing frameworks, groups can carry out desk operations instantly from Python purposes, making it appropriate for small to medium-scale information exploration and evaluation with minimal studying curve. As well as, PyIceberg is built-in with Python information evaluation libraries like Pandas and Polars, so information customers can use their present abilities and workflows.

When utilizing PyIceberg with the Knowledge Catalog and Amazon Easy Storage Service (Amazon S3), information groups can retailer and handle their tables in a totally serverless atmosphere. This implies information groups can deal with evaluation and insights somewhat than infrastructure administration.

Moreover, Iceberg tables managed by PyIceberg are suitable with AWS information analytics providers. Though PyIceberg operates on a single node and has efficiency limitations with giant information volumes, the identical tables might be effectively processed at scale utilizing providers reminiscent of Amazon Athena and AWS Glue. This allows groups to make use of PyIceberg for fast growth and testing, then transition to manufacturing workloads with larger-scale processing engines—whereas sustaining consistency of their information administration method.

Consultant use case

The next are widespread situations the place PyIceberg might be notably helpful:

• Knowledge science experimentation and have engineering – In information science, experiment reproducibility is essential for sustaining dependable and environment friendly analyses and fashions. Nonetheless, repeatedly updating organizational information makes it difficult to handle information snapshots for necessary enterprise occasions, mannequin coaching, and constant reference. Knowledge scientists can question historic snapshots by time journey capabilities and report necessary variations utilizing tagging options. With PyIceberg, they will obtain these advantages of their Python atmosphere utilizing acquainted instruments like Pandas. Due to Iceberg’s ACID capabilities, they will entry constant information even when tables are being actively up to date.
• Serverless information processing with Lambda – Organizations typically must course of information and keep analytical tables effectively with out managing advanced infrastructure. Utilizing PyIceberg with Lambda, groups can construct event-driven information processing and scheduled desk updates by serverless capabilities. PyIceberg’s light-weight nature makes it well-suited for serverless environments, enabling easy information processing duties like information validation, transformation, and ingestion. These tables stay accessible for each updates and analytics by numerous AWS providers, permitting groups to construct environment friendly information pipelines with out managing servers or clusters.

Occasion-driven information ingestion and evaluation with PyIceberg

On this part, we discover a sensible instance of utilizing PyIceberg for information processing and evaluation utilizing NYC yellow taxi journey information. To simulate an event-driven information processing state of affairs, we use Lambda to insert pattern information into an Iceberg desk, representing how real-time taxi journey information is likely to be processed. This instance will exhibit how PyIceberg can streamline workflows by combining environment friendly information ingestion with versatile evaluation capabilities.

Think about your group faces a number of necessities:

• The info processing answer must be cost-effective and maintainable, avoiding the complexity of managing distributed computing clusters for this moderately-sized dataset.
• Analysts want the flexibility to carry out versatile queries and explorations utilizing acquainted Python instruments. For instance, they could want to match historic snapshots with present information to research tendencies over time.
• The answer ought to have the flexibility to develop to be extra scalable sooner or later.

To deal with these necessities, we implement an answer that mixes Lambda for information processing with Jupyter notebooks for evaluation, each powered by PyIceberg. This method supplies a light-weight but strong structure that maintains information consistency whereas enabling versatile evaluation workflows. On the finish of the walkthrough, we additionally question this information utilizing Athena to exhibit compatibility with a number of Iceberg-supporting instruments and present how the structure can scale.

We stroll by the next high-level steps:

1. Use Lambda to write down pattern NYC yellow taxi journey information to an Iceberg desk on Amazon S3 utilizing PyIceberg with an AWS Glue Iceberg REST endpoint. In a real-world state of affairs, this Lambda perform can be triggered by an occasion from a queuing element like Amazon Easy Queue Service (Amazon SQS). For extra particulars, see Utilizing Lambda with Amazon SQS.
2. Analyze desk information in a Jupyter pocket book utilizing PyIceberg by the AWS Glue Iceberg REST endpoint.
3. Question the information utilizing Athena to exhibit Iceberg’s flexibility.

The next diagram illustrates the structure.

When implementing this structure, it’s necessary to notice that Lambda capabilities can have a number of concurrent invocations when triggered by occasions. This concurrent invocation would possibly result in transaction conflicts when writing to Iceberg tables. To deal with this, you need to implement an acceptable retry mechanism and punctiliously handle concurrency ranges. If you happen to’re utilizing Amazon SQS as an occasion supply, you may management concurrent invocations by the SQS occasion supply’s most concurrency setting.

Conditions

The next stipulations are essential for this use case:

Arrange assets with AWS CloudFormation

You should utilize the offered CloudFormation template to arrange the next assets:

Full the next steps to deploy the assets:

1. Select Launch stack.

2. For Parameters, pyiceberg_lambda_blog_database is about by default. You may also change the default worth. If you happen to change the database title, keep in mind to switch pyiceberg_lambda_blog_database together with your chosen title in all subsequent steps. Then, select Subsequent.
3. Select Subsequent.
4. Choose I acknowledge that AWS CloudFormation would possibly create IAM assets with customized names.
5. Select Submit.

Construct and run a Lambda perform

Let’s construct a Lambda perform to course of incoming information utilizing PyIceberg. This perform creates an Iceberg desk referred to as nyc_yellow_table within the database pyiceberg_lambda_blog_database within the Knowledge Catalog if it doesn’t exist. It then generates pattern NYC taxi journey information to simulate incoming information and inserts it into nyc_yellow_table.

Though we invoke this perform manually on this instance, in real-world situations, this Lambda perform can be triggered by precise occasions, reminiscent of messages from Amazon SQS. When implementing real-world use circumstances, the perform code should be modified to obtain the occasion information and course of it primarily based on the necessities.

We deploy the perform utilizing container photographs because the deployment package deal. To create a Lambda perform from a container picture, construct your picture on CloudShell and push it to an ECR repository. Full the next steps:

1. Check in to the AWS Administration Console and launch CloudShell.
2. Create a working listing.

mkdir pyiceberg_blog
cd pyiceberg_blog

3. Obtain the Lambda script lambda_function.py.

aws s3 cp s3://aws-blogs-artifacts-public/artifacts/BDB-5013/lambda_function.py .

This script performs the next duties:

• Creates an Iceberg desk with the NYC taxi schema within the Knowledge Catalog
• Generates a random NYC taxi dataset
• Inserts this information into the desk

Let’s break down the important elements of this Lambda perform:

• Iceberg catalog configuration – The next code defines an Iceberg catalog that connects to the AWS Glue Iceberg REST endpoint:

# Configure the catalog
catalog_properties = {
   "kind": "relaxation",
   "uri": f"https://glue.{area}.amazonaws.com/iceberg",
   "s3.area": area,
   "relaxation.sigv4-enabled": "true",
   "relaxation.signing-name": "glue",
   "relaxation.signing-region": area
}
catalog = load_catalog(**catalog_properties)

• Desk schema definition – The next code defines the Iceberg desk schema for the NYC taxi dataset. The desk consists of:
  • Schema columns outlined within the Schema
  • Partitioning by vendorid and tpep_pickup_datetime utilizing PartitionSpec
  • Day rework utilized to tpep_pickup_datetime for every day report administration
  • Type ordering by tpep_pickup_datetime and tpep_dropoff_datetime

When making use of the day rework to timestamp columns, Iceberg routinely handles date-based partitioning hierarchically. This implies a single day rework permits partition pruning on the 12 months, month, and day ranges with out requiring express transforms for every stage. For extra particulars about Iceberg partitioning, see Partitioning.

# Desk Definition
schema = Schema(
    NestedField(field_id=1, title="vendorid", field_type=LongType(), required=False),
    NestedField(field_id=2, title="tpep_pickup_datetime", field_type=TimestampType(), required=False),
    NestedField(field_id=3, title="tpep_dropoff_datetime", field_type=TimestampType(), required=False),
    NestedField(field_id=4, title="passenger_count", field_type=LongType(), required=False),
    NestedField(field_id=5, title="trip_distance", field_type=DoubleType(), required=False),
    NestedField(field_id=6, title="ratecodeid", field_type=LongType(), required=False),
    NestedField(field_id=7, title="store_and_fwd_flag", field_type=StringType(), required=False),
    NestedField(field_id=8, title="pulocationid", field_type=LongType(), required=False),
    NestedField(field_id=9, title="dolocationid", field_type=LongType(), required=False),
    NestedField(field_id=10, title="payment_type", field_type=LongType(), required=False),
    NestedField(field_id=11, title="fare_amount", field_type=DoubleType(), required=False),
    NestedField(field_id=12, title="additional", field_type=DoubleType(), required=False),
    NestedField(field_id=13, title="mta_tax", field_type=DoubleType(), required=False),
    NestedField(field_id=14, title="tip_amount", field_type=DoubleType(), required=False),
    NestedField(field_id=15, title="tolls_amount", field_type=DoubleType(), required=False),
    NestedField(field_id=16, title="improvement_surcharge", field_type=DoubleType(), required=False),
    NestedField(field_id=17, title="total_amount", field_type=DoubleType(), required=False),
    NestedField(field_id=18, title="congestion_surcharge", field_type=DoubleType(), required=False),
    NestedField(field_id=19, title="airport_fee", field_type=DoubleType(), required=False),
)

# Outline partition spec
partition_spec = PartitionSpec(
    PartitionField(source_id=1, field_id=1001, rework=IdentityTransform(), title="vendorid_idenitty"),
    PartitionField(source_id=2, field_id=1002, rework=DayTransform(), title="tpep_pickup_day"),
)

# Outline type order
sort_order = SortOrder(
    SortField(source_id=2, rework=DayTransform()),
    SortField(source_id=3, rework=DayTransform())
)

database_name = os.environ.get('GLUE_DATABASE_NAME')
table_name = os.environ.get('ICEBERG_TABLE_NAME')
identifier = f"{database_name}.{table_name}"

# Create the desk if it does not exist
location = f"s3://pyiceberg-lambda-blog-{account_id}-{area}/{database_name}/{table_name}"
if not catalog.table_exists(identifier):
    desk = catalog.create_table(
        identifier=identifier,
        schema=schema,
        location=location,
        partition_spec=partition_spec,
        sort_order=sort_order
    )
else:
    desk = catalog.load_table(identifier=identifier)

• Knowledge era and insertion – The next code generates random information and inserts it into the desk. This instance demonstrates an append-only sample, the place new information are repeatedly added to trace enterprise occasions and transactions:

# Generate random information
information = generate_random_data()
# Convert to Arrow Desk
df = pa.Desk.from_pylist(information)
# Write information utilizing PyIceberg
desk.append(df)

4. Obtain the Dockerfile. It defines the container picture in your perform code.

aws s3 cp s3://aws-blogs-artifacts-public/artifacts/BDB-5013/Dockerfile .

5. Obtain the necessities.txt. It defines the Python packages required in your perform code.

aws s3 cp s3://aws-blogs-artifacts-public/artifacts/BDB-5013/necessities.txt .

At this level, your working listing ought to comprise the next three information:

7. Set the atmosphere variables. Change <account_id> together with your AWS account ID:

export AWS_ACCOUNT_ID=<account_id>

8. Construct the Docker picture:

docker construct --provenance=false -t localhost/pyiceberg-lambda .

# Affirm constructed picture
docker photographs | grep pyiceberg-lambda

9. Set a tag to the picture:

docker tag localhost/pyiceberg-lambda:newest ${AWS_ACCOUNT_ID}.dkr.ecr.${AWS_REGION}.amazonaws.com/pyiceberg-lambda-repository:newest

10. Log in to the ECR repository created by AWS CloudFormation:

aws ecr get-login-password --region ${AWS_REGION} | docker login --username AWS --password-stdin ${AWS_ACCOUNT_ID}.dkr.ecr.${AWS_REGION}.amazonaws.com

11. Push the picture to the ECR repository:

docker push ${AWS_ACCOUNT_ID}.dkr.ecr.${AWS_REGION}.amazonaws.com/pyiceberg-lambda-repository:newest

12. Create a Lambda perform utilizing the container picture you pushed to Amazon ECR:

aws lambda create-function 
--function-name pyiceberg-lambda-function 
--package-type Picture 
--code ImageUri=${AWS_ACCOUNT_ID}.dkr.ecr.${AWS_REGION}.amazonaws.com/pyiceberg-lambda-repository:newest 
--role arn:aws:iam::${AWS_ACCOUNT_ID}:position/pyiceberg-lambda-function-role-${AWS_REGION} 
--environment "Variables={ICEBERG_TABLE_NAME=nyc_yellow_table, GLUE_DATABASE_NAME=pyiceberg_lambda_blog_database}" 
--region ${AWS_REGION} 
--timeout 60 
--memory-size 1024

13. Invoke the perform not less than 5 occasions to create a number of snapshots, which we’ll study within the following sections. Notice that we’re invoking the perform manually to simulate event-driven information ingestion. In actual world situations, Lambda capabilities will likely be routinely invoked with event-driven vogue.

aws lambda invoke 
--function-name arn:aws:lambda:${AWS_REGION}:${AWS_ACCOUNT_ID}:perform:pyiceberg-lambda-function 
--log-type Tail 
outputfile.txt 
--query 'LogResult' | tr -d '"' | base64 -d

At this level, you will have deployed and run the Lambda perform. The perform creates the nyc_yellow_table Iceberg desk within the pyiceberg_lambda_blog_database database. It additionally generates and inserts pattern information into this desk. We’ll discover the information within the desk in later steps.

For extra detailed details about constructing Lambda capabilities with containers, see Create a Lambda perform utilizing a container picture.

Discover the information with Jupyter utilizing PyIceberg

On this part, we exhibit learn how to entry and analyze the information saved in Iceberg tables registered within the Knowledge Catalog. Utilizing a Jupyter pocket book with PyIceberg, we entry the taxi journey information created by our Lambda perform and study totally different snapshots as new information arrive. We additionally tag particular snapshots to retain necessary ones, and create new tables for additional evaluation.

Full the next steps to open the pocket book with Jupyter on the SageMaker AI pocket book occasion:

1. On the SageMaker AI console, select Notebooks within the navigation pane.
2. Select Open JupyterLab subsequent to the pocket book that you just created utilizing the CloudFormation template.

3. Obtain the pocket book and open it in a Jupyter atmosphere in your SageMaker AI pocket book.

4. Open uploaded pyiceberg_notebook.ipynb.
5. Within the kernel choice dialog, depart the default choice and select Choose.

From this level ahead, you’ll work by the pocket book by operating cells so as.

Connecting Catalog and Scanning Tables

You’ll be able to entry the Iceberg desk utilizing PyIceberg. The next code connects to the AWS Glue Iceberg REST endpoint and hundreds the nyc_yellow_table desk on the pyiceberg_lambda_blog_database database:

import pyarrow as pa
from pyiceberg.catalog import load_catalog
import boto3

# Set AWS area
sts = boto3.shopper('sts')
area = sts._client_config.region_name

# Configure catalog connection properties
catalog_properties = {
    "kind": "relaxation",
    "uri": f"https://glue.{area}.amazonaws.com/iceberg",
    "s3.area": area,
    "relaxation.sigv4-enabled": "true",
    "relaxation.signing-name": "glue",
    "relaxation.signing-region": area
}

# Specify database and desk names
database_name = "pyiceberg_lambda_blog_database"
table_name = "nyc_yellow_table"

# Load catalog and get desk
catalog = load_catalog(**catalog_properties)
desk = catalog.load_table(f"{database_name}.{table_name}")

You’ll be able to question full information from the Iceberg desk as an Apache Arrow desk and convert it to a Pandas DataFrame.

Working with Snapshots

One of many necessary options of Iceberg is snapshot-based model management. Snapshots are routinely created each time information adjustments happen within the desk. You’ll be able to retrieve information from a selected snapshot, as proven within the following instance.

# Get information from a selected snapshot ID
snapshot_id = snapshots.to_pandas()["snapshot_id"][3]
snapshot_pa_table = desk.scan(snapshot_id=snapshot_id).to_arrow()
snapshot_df = snapshot_pa_table.to_pandas()

You’ll be able to examine the present information with historic information from any time limit primarily based on snapshots. On this case, you’re evaluating the variations in information distribution between the most recent desk and a snapshot desk:

# Evaluate the distribution of total_amount within the specified snapshot and the most recent information.
import matplotlib.pyplot as plt

plt.determine(figsize=(4, 3))
df['total_amount'].hist(bins=30, density=True, label="newest", alpha=0.5)
snapshot_df['total_amount'].hist(bins=30, density=True, label="snapshot", alpha=0.5)
plt.title('Distribution of total_amount')
plt.xlabel('total_amount')
plt.ylabel('relative Frequency')
plt.legend()
plt.present()

Tagging snapshots

You’ll be able to tag particular snapshots with an arbitrary title and question particular snapshots with that title later. That is helpful when managing snapshots of necessary occasions.

On this instance, you question a snapshot specifying the tag checkpointTag. Right here, you’re utilizing the polars to create a brand new DataFrame by including a brand new column referred to as trip_duration primarily based on present columns tpep_dropoff_datetime and tpep_pickup_datetime columns:

# retrive tagged snapshot desk as polars information body
import polars as pl

# Get snapshot id from tag title
df = desk.examine.refs().to_pandas()
filtered_df = df[df["name"] == tag_name]
tag_snapshot_id = filtered_df["snapshot_id"].iloc[0]

# Scan Desk primarily based on the snapshot id
tag_pa_table = desk.scan(snapshot_id=tag_snapshot_id).to_arrow()
tag_df = pl.from_arrow(tag_pa_table)

# Course of the information including a brand new column "trip_duration" from verify level snapshot.
def preprocess_data(df):
    df = df.choose(["vendorid", "tpep_pickup_datetime", "tpep_dropoff_datetime", 
                    "passenger_count", "trip_distance", "fare_amount"])
    df = df.with_columns(
        ((pl.col("tpep_dropoff_datetime") - pl.col("tpep_pickup_datetime"))
         .dt.total_seconds() // 60).alias("trip_duration"))
    return df

processed_df = preprocess_data(tag_df)
show(processed_df)
print(processed_df["trip_duration"].describe())

Create a brand new desk from the processed DataFrame with the trip_duration column. This step illustrates learn how to put together information for potential future evaluation. You’ll be able to explicitly specify the snapshot of the information that the processed information is referring to by utilizing a tag, even when the underlying desk has been modified.

# write processed information to new iceberg desk
account_id = sts.get_caller_identity()["Account"] 

new_table_name = "processed_" + table_name
location = f"s3://pyiceberg-lambda-blog-{account_id}-{area}/{database_name}/{new_table_name}"

pa_new_table = processed_df.to_arrow()
schema = pa_new_table.schema
identifier = f"{database_name}.{new_table_name}"

new_table = catalog.create_table(
                identifier=identifier,
                schema=schema,
                location=location
            )
            
# present new desk's schema
print(new_table.schema())
# insert processed information to new desk
new_table.append(pa_new_table)

Let’s question this new desk created from processed information with Athena to exhibit the Iceberg desk’s interoperability.

Question the information from Athena

1. Within the Athena question editor, you may question the desk pyiceberg_lambda_blog_database.processed_nyc_yellow_table created from the pocket book within the earlier part:

SELECT * FROM "pyiceberg_lambda_blog_database"."processed_nyc_yellow_table" restrict 10;

By finishing these steps, you’ve constructed a serverless information processing answer utilizing PyIceberg with Lambda and an AWS Glue Iceberg REST endpoint. You’ve labored with PyIceberg to handle and analyze information utilizing Python, together with snapshot administration and desk operations. As well as, you ran the question utilizing one other engine, Athena, which exhibits the compatibility of the Iceberg desk.

Clear up

To wash up the assets used on this submit, full the next steps:

1. On the Amazon ECR console, navigate to the repository pyiceberg-lambda-repository and delete all photographs contained within the repository.
2. On the CloudShell, delete working listing pyiceberg_blog.
3. On the Amazon S3 console, navigate to the S3 bucket pyiceberg-lambda-blog-<ACCOUNT_ID>-<REGION>, which you created utilizing the CloudFormation template, and empty the bucket.
4. After you affirm the repository and the bucket are empty, delete the CloudFormation stack pyiceberg-lambda-blog-stack.
5. Delete the Lambda perform pyiceberg-lambda-function that you just created utilizing the Docker picture.

Conclusion

On this submit, we demonstrated how utilizing PyIceberg with the AWS Glue Knowledge Catalog permits environment friendly, light-weight information workflows whereas sustaining strong information administration capabilities. We showcased how groups can use Iceberg’s highly effective options with minimal setup and infrastructure dependencies. This method permits organizations to start out working with Iceberg tables shortly, with out the complexity of establishing and managing distributed computing assets.

That is notably priceless for organizations trying to undertake Iceberg’s capabilities with a low barrier to entry. The light-weight nature of PyIceberg permits groups to start working with Iceberg tables instantly, utilizing acquainted instruments and requiring minimal further studying. As information wants develop, the identical Iceberg tables might be seamlessly accessed by AWS analytics providers like Athena and AWS Glue, offering a transparent path for future scalability.

To study extra about PyIceberg and AWS analytics providers, we encourage you to discover the PyIceberg documentation and What’s Apache Iceberg?

Concerning the authors

Sotaro Hikita is a Specialist Options Architect targeted on analytics with AWS, working with large information applied sciences and open supply software program. Exterior of labor, he at all times seeks out good meals and has just lately change into obsessed with pizza.

Shuhei Fukami is a Specialist Options Architect targeted on Analytics with AWS. He likes cooking in his spare time and has change into obsessive about making pizza lately.