Run Apache XTable in AWS Lambda for background conversion of open desk codecs

This put up was co-written with Dipankar Mazumdar, Employees Information Engineering Advocate with AWS Associate OneHouse.

Information structure has developed considerably to deal with rising knowledge volumes and numerous workloads. Initially, knowledge warehouses had been the go-to answer for structured knowledge and analytical workloads however had been restricted by proprietary storage codecs and their incapacity to deal with unstructured knowledge. This led to the rise of knowledge lakes primarily based on columnar codecs like Apache Parquet, which got here with totally different challenges like the shortage of ACID capabilities.

Finally, transactional knowledge lakes emerged so as to add transactional consistency and efficiency of an information warehouse to the information lake. Central to a transactional knowledge lake are open desk codecs (OTFs) equivalent to Apache Hudi, Apache Iceberg, and Delta Lake, which act as a metadata layer over columnar codecs. These codecs present important options like schema evolution, partitioning, ACID transactions, and time-travel capabilities, that deal with conventional issues in knowledge lakes.

In apply, OTFs are utilized in a broad vary of analytical workloads, from enterprise intelligence to machine studying. Furthermore, they are often mixed to learn from particular person strengths. For example, a streaming knowledge pipeline can write tables utilizing Hudi due to its power in low-latency, write-heavy workloads. In later pipeline phases, knowledge is transformed to Iceberg, to learn from its learn efficiency. Historically, this conversion required time-consuming rewrites of knowledge recordsdata, leading to knowledge duplication, increased storage, and elevated compute prices. In response, the business is shifting towards interoperability between OTFs, with instruments that permit conversions with out knowledge duplication. Apache XTable (incubating), an rising open supply undertaking, facilitates seamless conversions between OTFs, eliminating lots of the challenges related to desk format conversion.

On this put up, we discover how Apache XTable, mixed with the AWS Glue Information Catalog, permits background conversions between OTFs residing on Amazon Easy Storage Service (Amazon S3) primarily based knowledge lakes, with minimal to no modifications to current pipelines in a scalable and cost-effective manner, as proven within the following diagram.

This put up is considered one of a number of posts about XTable on AWS. For extra examples and references to different posts, discuss with the next GitHub repository.

Apache XTable

Apache XTable (incubating) is an open supply undertaking designed to allow interoperability amongst varied knowledge lake desk codecs, permitting omnidirectional conversions between codecs with out the necessity to copy or rewrite knowledge. Initially open sourced in November 2023 below the title OneTable, with contributions from amongst others OneHouse, it was licensed below Apache 2.0. In March 2024, the undertaking was donated to the Apache Software program Basis (ASF) and rebranded as Apache XTable, the place it’s now incubating. XTable isn’t a brand new desk format however supplies abstractions and instruments to translate the metadata related to current codecs. The first goal of XTable is to permit customers to begin with any desk format and have the flexibleness to change to a different as wanted.

Inside workings and options

At a basic stage, Hudi, Iceberg, and Delta Lake share similarities of their construction. When knowledge is written to a distributed file system, these codecs include an information layer, usually Parquet recordsdata, and a metadata layer that gives the required abstraction (see the next diagram). XTable makes use of these commonalities to allow interoperability between codecs.

The synchronization course of in XTable works by translating desk metadata utilizing the present APIs of those desk codecs. It reads the present metadata from the supply desk and generates the corresponding metadata for a number of goal codecs. This metadata is then saved in a delegated listing inside the base path of your desk, equivalent to _delta_log for Delta Lake, metadata for Iceberg, and .hoodie for Hudi. This enables the present knowledge to be interpreted as if it had been initially written in any of those codecs.

XTable supplies two metadata translation strategies: Full Sync, which interprets all commits, and Incremental Sync, which solely interprets new, unsynced commits for larger effectivity with massive tables. If points come up with Incremental Sync, XTable routinely falls again to Full Sync to offer uninterrupted translation.

Neighborhood and future

When it comes to future plans, XTable is targeted on attaining characteristic parity with OTFs’ built-in options, together with including vital capabilities like assist for Merge-on-Learn (MoR) tables. The undertaking additionally plans to facilitate synchronization of desk codecs throughout a number of catalogs, equivalent to AWS Glue, Hive, and Unity catalog.

Run XTable as a steady background conversion mechanism

On this put up, we describe a background conversion mechanism for OTFs that doesn’t require modifications to knowledge pipelines. The mechanism periodically scans an information catalog just like the AWS Glue Information Catalog for tables to transform with XTable.

On an information platform, an information catalog shops desk metadata and usually accommodates the information mannequin and bodily storage location of the datasets. It serves because the central integration with analytical providers. To maximise ease of use, compatibility, and scalability on AWS, the conversion mechanism described on this put up is constructed across the AWS Glue Information Catalog.

The next diagram illustrates the answer at a look. We design this conversion mechanism primarily based on Lambda, AWS Glue, and XTable.

To ensure that the Lambda perform to have the ability to detect the tables contained in the Information Catalog, the next data must be related to a desk: supply format and goal codecs. For every detected desk, the Lambda perform invokes the XTable software, which is packaged into the capabilities surroundings. Then XTable interprets between supply and goal codecs and writes the brand new metadata on the identical knowledge retailer.

Answer overview

We implement the answer with the AWS Cloud Growth Package (AWS CDK), an open supply software program growth framework for outlining cloud infrastructure in code, and supply it on GitHub. The AWS CDK answer deploys the next parts:

• A converter Lambda perform that accommodates the XTable software and begins the conversion job for the detected tables
• A detector Lambda perform that scans the Information Catalog for tables which can be to be transformed and invokes the converter Lambda perform
• An Amazon EventBridge schedule that invokes the detector Lambda perform on an hourly foundation

At the moment, the XTable software must be constructed from supply. We subsequently present a Dockerfile that implements the required construct steps and use the ensuing Docker picture because the Lambda perform runtime surroundings.

In case you don’t have pattern knowledge obtainable for testing, we offer scripts for producing pattern datasets on GitHub. Information and metadata are proven in blue within the following element diagram.

Converter Lambda perform: Run XTable

The converter Lambda perform invokes the XTable JAR, wrapped with the third-party library jpype, and converts the metadata layer of the respective knowledge lake tables.

The perform is outlined within the AWS CDK by way of the DockerImageFunction, which makes use of a Dockerfile and builds a Docker container as a part of the deploy step. With this mechanism, we will bundle the XTable software inside our Lambda perform.

First, we obtain the XTtable GitHub repository and construct the jar with the maven CLI. That is performed as part of the Docker container construct course of:

# Dockerfile # clone sources
RUN git clone --depth 1 --branch <xtable_branch> https://github.com/apache/incubator-xtable.git

# construct xtable jar
WORKDIR /incubator-xtable
RUN /apache-maven-<maven_version>/bin/mvn package deal -DskipTests=true
WORKDIR /

To routinely construct and add the Docker picture, we create a DockerImageFunction within the AWS CDK and reference the Dockerfile in its definition. To efficiently run Spark and subsequently XTable in a Lambda perform, we have to set the LOCAL_IP variable of Spark to localhost and subsequently to 127.0.0.1:

# cdk_stack.py
detector = _lambda.DockerImageFunction(
    scope=self,
    id="Converter",
    # Dockerfile in ./src listing
    code=_lambda.DockerImageCode.from_image_asset(
        listing="src", cmd=["detector.handler"]
    )
    surroundings={"SPARK_LOCAL_IP": "127.0.0.1"}
    ...
)

To name the XTtable JAR, we use a third-party Python library referred to as jpype, which handles the communication with the Java digital machine. In our Python code, the XTtable name is as follows:

# name java class with configuration recordsdata
run_sync = jpype.JPackage("org").apache.xtable.utilities.RunSync.fundamental
run_sync(
    [
        "--datasetConfig",
        "<path_to_dataset_config>",
        "--icebergCatalogConfig",
        "<path_to_catalog_config>",
    ]
)

For extra data on XTable software parameters, see Creating your first interoperable desk.

Detector Lambda perform: Determine tables to transform within the Information Catalog

The detector Lambda perform scans the tables within the Information Catalog. For a desk that might be transformed, it invokes the converter Lambda perform by way of an occasion. This decouples the scanning and conversion components and makes our answer extra resilient to potential failures.

The detection mechanism searches within the desk parameters for the parameters xtable_table_type and xtable_target_formats. In the event that they exist, the conversion is invoked. See the next code:

# detector.py
# create paginator to loop by way of AWS Glue tables
tables = glue_client.get_paginator("get_tables").paginate(
    DatabaseName=database["Name"]
)
for table_list in tables:
    table_list = table_list["TableList"]
…
# loop by way of all tables and examine for required customized glue parameters
for desk in table_list:
    required_parameters={"xtable_table_type", "xtable_target_formats"}
    # if required desk parameters exist cross on desk for conversion
    if required_parameters <= desk["Parameters"].keys():
        yield desk

EventBridge Scheduler rule

Within the AWS CDK, you outline an EventBridge Scheduler rule as follows. Based mostly on the rule, EventBridge will then name the Lambda detector perform each hour:

# cdk_stack.py
occasion = occasions.Rule(
    scope=self,
    id="DetectorSchedule",
    schedule=occasions.Schedule.fee(Period.hours(1)),
)
occasion.add_target(targets.LambdaFunction(detector))

Stipulations

Let’s dive deeper into the right way to deploy the offered AWS CDK stack. You want one of many following container runtimes:

• Finch (an open supply consumer for container growth)
• Docker

You additionally want the AWS CDK configured. For extra particulars, see Getting began with the AWS CDK.

Construct and deploy the answer

Full the next steps:

1. To deploy the stack, clone the GitHub repo, develop into the folder for this put up (xtable_lambda), and deploy the AWS CDK stack:

   git clone https://github.com/aws-samples/apache-xtable-on-aws-samples.git
   cd xtable_lambda
   cdk deploy

This deploys the described Lambda capabilities and the EventBridge Scheduler rule.

2. When utilizing Finch, it’s essential set the CDK_DOCKER surroundings variable earlier than deployment:

After profitable deployment, the conversion mechanism begins to run each hour.

3. The next parameters have to exist on the AWS Glue desk that might be transformed:
   1. "xtable_table_type": "<source_format>"
   2. "xtable_target_formats": "<target_format>, <target_format>"

On the AWS Glue console, the parameters seem like the next screenshot and could be set below Desk properties when enhancing an AWS Glue desk.

4. Optionally, for those who don’t have pattern knowledge, the next scripts can assist you arrange a take a look at surroundings both together with your native machine or in an AWS Glue for Spark job:

   # native: create hudi dataset on S3
   cd scripts
   pip set up -r necessities.txt
   python ./create_hudi_s3.py

Convert a streaming desk (Hudi to Iceberg)

Let’s assume we now have a Hudi desk on Amazon S3, which is registered within the Information Catalog, and wish to periodically translate it to Iceberg format. Information is streaming in constantly. We’ve deployed the offered AWS CDK stack and set the required AWS Glue desk properties to translate the dataset to the Iceberg format. Within the following steps, we run the background job, see the leads to AWS Glue and Amazon S3, and question it with Amazon Athena, a serverless and interactive analytics service that gives a simplified and versatile technique to analyze petabytes of knowledge.

In Amazon S3 and AWS Glue, we will see our Hudi dataset and desk together with the metadata folder .hoodie. On the AWS Glue console, we set the next desk properties:

• "xtable_target_type": "HUDI"
• "xtable_table_formats": "ICEBERG"

Our Lambda perform is invoked periodically each hour. After the run, we will discover the Iceberg-specific metadata folder in our S3 bucket, which was generated by XTable.

If we have a look at the Information Catalog, we will see the brand new desk <table_name>_converted was registered as an Iceberg desk.

With the Iceberg format, we will now reap the benefits of the time journey characteristic by querying the dataset with a downstream analytical service like Athena. Within the following screenshot, you’ll be able to see at Title: that the desk is in Iceberg format.

Querying all snapshots, we will see that we created three snapshots with overwrites after the preliminary one.

We then take the present time and question the dataset illustration of 180 minutes in the past, ensuing within the knowledge from the primary snapshot dedicated.

Abstract

On this put up, we demonstrated the right way to construct a background conversion job for OTFs, utilizing XTable and the Information Catalog, which is unbiased from knowledge pipelines and transformation jobs. By way of Xtable, it permits for environment friendly translation between OTFs, as a result of knowledge recordsdata are reused and solely the metadata layer is processed. The mixing with the Information Catalog supplies large compatability with AWS analytical providers.

You possibly can reuse the Lambda primarily based XTable deployment in different options. For example, you would use it in a reactive mechanism for close to real-time conversion of OTFs, which is invoked by Amazon S3 object occasions ensuing from modifications to OTF metadata.

For additional details about XTable, see the undertaking’s official web site. For extra examples and references to different posts on utilizing XTable on AWS, discuss with the next GitHub repository.

Concerning the authors

Matthias Rudolph is a Options Architect at AWS, digitalizing the German manufacturing business, specializing in analytics and large knowledge. Earlier than that he was a lead developer on the German producer KraussMaffei Applied sciences, liable for the event of knowledge platforms.

Dipankar Mazumdar is a Employees Information Engineer Advocate at Onehouse.ai, specializing in open-source tasks like Apache Hudi and XTable to assist engineering groups construct and scale sturdy analytics platforms, with prior contributions to vital tasks equivalent to Apache Iceberg and Apache Arrow.

Stephen Stated is a Senior Options Architect and works with Retail/CPG clients. His areas of curiosity are knowledge platforms and cloud-native software program engineering.