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 instances turn out to be extra complicated, information engineering groups require refined tooling to deal with versioning, growing 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 will be accessed from varied distributed information processing frameworks like Apache Spark and Trino, making it a versatile resolution for numerous information processing wants. Among the many accessible 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 put up, we display how PyIceberg, built-in with the AWS Glue Information Catalog and AWS Lambda, supplies a light-weight strategy 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

Considered one of PyIceberg’s major 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 expertise and workflows.

When utilizing PyIceberg with the Information 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 concentrate on evaluation and insights fairly than infrastructure administration.

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

Consultant use case

The next are frequent situations the place PyIceberg will be notably helpful:

• Information science experimentation and have engineering – In information science, experiment reproducibility is essential for sustaining dependable and environment friendly analyses and fashions. Nevertheless, constantly updating organizational information makes it difficult to handle information snapshots for necessary enterprise occasions, mannequin coaching, and constant reference. Information scientists can question historic snapshots by time journey capabilities and report necessary variations utilizing tagging options. With PyIceberg, they’ll obtain these advantages of their Python atmosphere utilizing acquainted instruments like Pandas. Due to Iceberg’s ACID capabilities, they’ll 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 complicated infrastructure. Utilizing PyIceberg with Lambda, groups can construct event-driven information processing and scheduled desk updates by serverless features. 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 varied AWS companies, 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 data could be processed. This instance will display 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 resolution 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 may want to check historic snapshots with present information to investigate developments over time.
• The answer ought to have the flexibility to broaden to be extra scalable sooner or later.

To handle these necessities, we implement an answer that mixes Lambda for information processing with Jupyter notebooks for evaluation, each powered by PyIceberg. This strategy supplies a light-weight but sturdy structure that maintains information consistency whereas enabling versatile evaluation workflows. On the finish of the walkthrough, we additionally question this information utilizing Athena to display 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 jot 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 could be triggered by an occasion from a queuing part 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 info utilizing Athena to display Iceberg’s flexibility.

The next diagram illustrates the structure.

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

Stipulations

The next conditions are obligatory 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 identify, keep in mind to switch pyiceberg_lambda_blog_database together with your chosen identify in all subsequent steps. Then, select Subsequent.
3. Select Subsequent.
4. Choose I acknowledge that AWS CloudFormation may 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 data utilizing PyIceberg. This perform creates an Iceberg desk referred to as nyc_yellow_table within the database pyiceberg_lambda_blog_database within the Information Catalog if it doesn’t exist. It then generates pattern NYC taxi journey information to simulate incoming data and inserts it into nyc_yellow_table.

Though we invoke this perform manually on this instance, in real-world situations, this Lambda perform could be triggered by precise occasions, equivalent to messages from Amazon SQS. When implementing real-world use instances, 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 bundle. 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. Register 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 Information 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 = {
   "sort": "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
  • Kind 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 yr, 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, identify="vendorid", field_type=LongType(), required=False),
    NestedField(field_id=2, identify="tpep_pickup_datetime", field_type=TimestampType(), required=False),
    NestedField(field_id=3, identify="tpep_dropoff_datetime", field_type=TimestampType(), required=False),
    NestedField(field_id=4, identify="passenger_count", field_type=LongType(), required=False),
    NestedField(field_id=5, identify="trip_distance", field_type=DoubleType(), required=False),
    NestedField(field_id=6, identify="ratecodeid", field_type=LongType(), required=False),
    NestedField(field_id=7, identify="store_and_fwd_flag", field_type=StringType(), required=False),
    NestedField(field_id=8, identify="pulocationid", field_type=LongType(), required=False),
    NestedField(field_id=9, identify="dolocationid", field_type=LongType(), required=False),
    NestedField(field_id=10, identify="payment_type", field_type=LongType(), required=False),
    NestedField(field_id=11, identify="fare_amount", field_type=DoubleType(), required=False),
    NestedField(field_id=12, identify="additional", field_type=DoubleType(), required=False),
    NestedField(field_id=13, identify="mta_tax", field_type=DoubleType(), required=False),
    NestedField(field_id=14, identify="tip_amount", field_type=DoubleType(), required=False),
    NestedField(field_id=15, identify="tolls_amount", field_type=DoubleType(), required=False),
    NestedField(field_id=16, identify="improvement_surcharge", field_type=DoubleType(), required=False),
    NestedField(field_id=17, identify="total_amount", field_type=DoubleType(), required=False),
    NestedField(field_id=18, identify="congestion_surcharge", field_type=DoubleType(), required=False),
    NestedField(field_id=19, identify="airport_fee", field_type=DoubleType(), required=False),
)

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

# Outline kind 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 would 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)

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

# Generate random information
data = generate_random_data()
# Convert to Arrow Desk
df = pa.Desk.from_pylist(data)
# 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 recordsdata:

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

8. Construct the Docker picture:

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

# Verify 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}:function/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 no less than 5 occasions to create a number of snapshots, which we’ll study within the following sections. Be aware that we’re invoking the perform manually to simulate event-driven information ingestion. In actual world situations, Lambda features shall 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 might 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 are going to discover the data within the desk in later steps.

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

Discover the info with Jupyter utilizing PyIceberg

On this part, we display the best way to entry and analyze the info saved in Iceberg tables registered within the Information Catalog. Utilizing a Jupyter pocket book with PyIceberg, we entry the taxi journey information created by our Lambda perform and study completely different snapshots as new data 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 possibility 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 masses 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 = {
    "sort": "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 at any time when information adjustments happen within the desk. You’ll be able to retrieve information from a particular snapshot, as proven within the following instance.

# Get information from a particular 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 evaluate the present information with historic information from any cut-off date primarily based on snapshots. On this case, you might be evaluating the variations in information distribution between the newest desk and a snapshot desk:

# Examine the distribution of total_amount within the specified snapshot and the newest 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 identify and question particular snapshots with that identify later. That is helpful when managing snapshots of necessary occasions.

On this instance, you question a snapshot specifying the tag checkpointTag. Right here, you might be 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 identify
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 info 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 the best way to put together information for potential future evaluation. You’ll be able to explicitly specify the snapshot of the info 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 constituted of processed information with Athena to display the Iceberg desk’s interoperability.

Question the info from Athena

1. Within the Athena question editor, you’ll be able to 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 resolution 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 scrub up the assets used on this put up, 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--, 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 put up, we demonstrated how utilizing PyIceberg with the AWS Glue Information Catalog permits environment friendly, light-weight information workflows whereas sustaining sturdy information administration capabilities. We showcased how groups can use Iceberg’s highly effective options with minimal setup and infrastructure dependencies. This strategy permits organizations to start out working with Iceberg tables shortly, with out the complexity of establishing and managing distributed computing assets.

That is notably useful for organizations seeking 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 will be seamlessly accessed by AWS analytics companies like Athena and AWS Glue, offering a transparent path for future scalability.

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

In regards to the authors

Sotaro Hikita is a Specialist Options Architect centered on analytics with AWS, working with massive information applied sciences and open supply software program. Outdoors of labor, he at all times seeks out good meals and has lately turn out to be obsessed with pizza.

Shuhei Fukami is a Specialist Options Architect centered on Analytics with AWS. He likes cooking in his spare time and has turn out to be obsessive about making pizza nowadays.