Learn how to build a near real-time transactional data lake on AWS using a combination of Open Source Software (OSS) and AWS Services

Introduction

In the following post, we will explore one possible architecture for building a near real-time transactional data lake on AWS. The data lake will be built using a combination of open source software (OSS) and fully-managed AWS services. Red Hat’s Debezium, Apache Kafka, and Kafka Connect will be used for change data capture (CDC). In addition, Apache Spark, Apache Hudi, and Hudi’s DeltaStreamer will be used to manage the data lake. To complete our architecture, we will use several fully-managed AWS services, including Amazon RDS, Amazon MKS, Amazon EKS, AWS Glue, and Amazon EMR.

Source Code

The source code, configuration files, and a list of commands shown in this post are open-sourced and available on GitHub.

Kafka

According to the Apache Kafka documentation, “Apache Kafka is an open-source distributed event streaming platform used by thousands of companies for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications.” For this post, we will use Apache Kafka as the core of our change data capture (CDC) process. According to Wikipedia, “change data capture (CDC) is a set of software design patterns used to determine and track the data that has changed so that action can be taken using the changed data. CDC is an approach to data integration that is based on the identification, capture, and delivery of the changes made to enterprise data sources.” We will discuss CDC in greater detail later in the post.

There are several options for Apache Kafka on AWS. We will use AWS’s fully-managed Amazon Managed Streaming for Apache Kafka (Amazon MSK) service. Alternatively, you could choose industry-leading SaaS providers, such as Confluent, Aiven, Redpanda, or Instaclustr. Lastly, you could choose to self-manage Apache Kafka on Amazon Elastic Compute Cloud (Amazon EC2) or Amazon Elastic Kubernetes Service (Amazon EKS).

Kafka Connect

According to the Apache Kafka documentation, “Kafka Connect is a tool for scalably and reliably streaming data between Apache Kafka and other systems. It makes it simple to quickly define connectors that move large collections of data into and out of Kafka. Kafka Connect can ingest entire databases or collect metrics from all your application servers into Kafka topics, making the data available for stream processing with low latency.”

There are multiple options for Kafka Connect on AWS. You can use AWS’s fully-managed, serverless Amazon MSK Connect. Alternatively, you could choose a SaaS provider or self-manage Kafka Connect yourself on Amazon Elastic Compute Cloud (Amazon EC2) or Amazon Elastic Kubernetes Service (Amazon EKS). I am not a huge fan of Amazon MSK Connect during development. In my opinion, iterating on the configuration for a new source and sink connector, especially with transforms and external registry dependencies, can be painfully slow and time-consuming with MSK Connect. I find it much faster to develop and fine-tune my sink and source connectors using a self-managed version of Kafka Connect. For Production workloads, you can easily port the configuration from the native Kafka Connect connector to MSK Connect. I am using a self-managed version of Kafka Connect for this post, running in Amazon EKS.

Debezium

According to the Debezium documentation, “Debezium is an open source distributed platform for change data capture. Start it up, point it at your databases, and your apps can start responding to all of the inserts, updates, and deletes that other apps commit to your databases.” Regarding Kafka Connect, according to the Debezium documentation, “Debezium is built on top of Apache Kafka and provides a set of Kafka Connect compatible connectors. Each of the connectors works with a specific database management system (DBMS). Connectors record the history of data changes in the DBMS by detecting changes as they occur and streaming a record of each change event to a Kafka topic. Consuming applications can then read the resulting event records from the Kafka topic.”

Source Connectors

We will use Kafka Connect along with three Debezium connectors, MySQL, PostgreSQL, and SQL Server, to connect to three corresponding Amazon Relational Database Service (RDS) databases and perform CDC. Changes from the three databases will be represented as messages in separate Kafka topics. In Kafka Connect terminology, these are referred to as Source Connectors. According to Confluent.io, a leader in the Kafka community, “source connectors ingest entire databases and stream table updates to Kafka topics.”

Sink Connector

We will stream the data from the Kafka topics into Amazon S3 using a sink connector. Again, according to Confluent.io, “sink connectors deliver data from Kafka topics to secondary indexes, such as Elasticsearch, or batch systems such as Hadoop for offline analysis.” We will use Confluent’s Amazon S3 Sink Connector for Confluent Platform. We can use Confluent’s sink connector without depending on the entire Confluent platform.

There is also an option to use the Hudi Sink Connector for Kafka, which promises to greatly simplify the processes described in this post. However, the RFC for this Hudi feature appears to be stalled. Last updated in August 2021, the RFC is still in the initial “Under Discussion” phase. Therefore, I would not recommend the connectors use in Production until it is GA (General Availability) and gains broader community support.

Securing Database Credentials

Whether using Amazon MSK Connect or self-managed Kafka Connect, you should ensure your database, Kafka, and schema registry credentials, and other sensitive configuration values are secured. Both MSK Connect and self-managed Kafka Connect can integrate with configuration providers that implement the ConfigProvider class interface, such as AWS Secrets Manager, HashiCorp Vault, Microsoft Azure Key Vault, and Google Cloud Secrets Manager.

For self-managed Kafka Connect, I prefer Jeremy Custenborder’s kafka-config-provider-aws plugin. This plugin provides integration with AWS Secrets Manager. A complete list of Jeremy’s providers can be found on GitHub. Below is a snippet of the Secrets Manager configuration from the connect-distributed.properties files, which is read by Apache Kafka Connect at startup.

https://garystafford.medium.com/media/db6ff589c946fcd014546c56d0255fe5

Apache Avro

The message in the Kafka topic and corresponding objects in Amazon S3 will be stored in Apache Avro format by the CDC process. The Apache Avro documentation states, “Apache Avro is the leading serialization format for record data, and the first choice for streaming data pipelines.” Avro provides rich data structures and a compact, fast, binary data format.

Again, according to the Apache Avro documentation, “Avro relies on schemas. When Avro data is read, the schema used when writing it is always present. This permits each datum to be written with no per-value overheads, making serialization both fast and small. This also facilitates use with dynamic, scripting languages, since data, together with its schema, is fully self-describing.” When Avro data is stored in a file, its schema can be stored with it so any program may process files later.

Alternatively, the schema can be stored separately in a schema registry. According to Apicurio Registry, “in the messaging and event streaming world, data that are published to topics and queues often must be serialized or validated using a Schema (e.g. Apache Avro, JSON Schema, or Google protocol buffers). Schemas can be packaged in each application, but it is often a better architectural pattern to instead register them in an external system [schema registry] and then referenced from each application.”

Schema Registry

Several leading open-source and commercial schema registries exist, including Confluent Schema Registry, AWS Glue Schema Registry, and Red Hat’s open-source Apicurio Registry. In this post, we use a self-managed version of Apicurio Registry running on Amazon EKS. You can relatively easily substitute AWS Glue Schema Registry if you prefer a fully-managed AWS service.

Apicurio Registry

According to the documentation, Apicurio Registry supports adding, removing, and updating OpenAPI, AsyncAPI, GraphQL, Apache Avro, Google protocol buffers, JSON Schema, Kafka Connect schema, WSDL, and XML Schema (XSD) artifact types. Furthermore, content evolution can be controlled by enabling content rules, including validity and compatibility. Lastly, the registry can be configured to store data in various backend storage systems depending on the use case, including Kafka (e.g., Amazon MSK), PostgreSQL (e.g., Amazon RDS), and Infinispan (embedded).

Data Lake Table Formats

Three leading open-source, transactional data lake storage frameworks enable building data lake and data lakehouse architectures: Apache Iceberg, Linux Foundation Delta Lake, and Apache Hudi. They offer comparable features, such as ACID-compliant transactional guarantees, time travel, rollback, and schema evolution. Using any of these three data lake table formats will allow us to store and perform analytics on the latest version of the data in the data lake originating from the three Amazon RDS databases.

Apache Hudi

According to the Apache Hudi documentation, “Apache Hudi is a transactional data lake platform that brings database and data warehouse capabilities to the data lake.” The specifics of how the data is laid out as files in your data lake depends on the Hudi table type you choose, either Copy on Write (CoW) or Merge On Read (MoR).

Like Apache Iceberg and Delta Lake, Apache Hudi is partially supported by AWS’s Analytics services, including AWS Glue, Amazon EMR, Amazon Athena, Amazon Redshift Spectrum, and AWS Lake Formation. In general, CoW has broader support on AWS than MoR. It is important to understand the limitations of Apache Hudi with each AWS analytics service before choosing a table format.

If you are looking for a fully-managed Cloud data lake service built on Hudi, I recommend Onehouse. Born from the roots of Apache Hudi and founded by its original creator, Vinoth Chandar (PMC chair of the Apache Hudi project), the Onehouse product and its services leverage OSS Hudi to offer a data lake platform similar to what companies like Uber have built.

Hudi DeltaStreamer

Hudi also offers DeltaStreamer (aka HoodieDeltaStreamer) for streaming ingestion of CDC data. DeltaStreamer can be run once or continuously, using Apache Spark, similar to an Apache Spark Structured Streaming job, to capture changes to the raw CDC data and write that data to a different part of our data lake.

DeltaStreamer also works with Amazon S3 Event Notifications instead of running continuously. According to DeltaStreamer’s documentation, Amazon S3 object storage provides an event notification service to post notifications when certain events happen in your S3 bucket. AWS will put these events in Amazon Simple Queue Service (Amazon SQS). Apache Hudi provides an S3EventsSource that can read from Amazon SQS to trigger and process new or changed data as soon as it is available on Amazon S3.

Sample Data for the Data Lake

The data used in this post is from the TICKIT sample database. The TICKIT database represents the backend data store for a platform that brings buyers and sellers of tickets to entertainment events together. It was initially designed to demonstrate Amazon Redshift. The TICKIT database is a small database with approximately 425K rows of data across seven tables: Category, Event, Venue, User, Listing, Sale, and Date.

A data lake most often contains data from multiple sources, each with different storage formats, protocols, and connection methods. To simulate these data sources, I have separated the TICKIT database tables to represent three typical enterprise systems, including a commercial off-the-shelf (COTS) E-commerce platform, a custom Customer Relationship Management (CRM) platform, and a SaaS-based Event Management System (EMS). Each simulated system uses a different Amazon RDS database engine, including MySQL, PostgreSQL, and SQL Server.

Enabling CDC for Amazon RDS

To use Debezium for CDC with Amazon RDS databases, minor changes to the default database configuration for each database engine are required.

CDC for PostgreSQL

Debezium has detailed instructions regarding configuring CDC for Amazon RDS for PostgreSQL. Database parameters specify how the database is configured. According to the Debezium documentation, for PostgreSQL, set the instance parameter rds.logical_replication to 1 and verify that the wal_level parameter is set to logical. It is automatically changed when the rds.logical_replication parameter is set to 1. This parameter is adjusted using an Amazon RDS custom parameter group. According to the AWS documentation, “with Amazon RDS, you manage database configuration by associating your DB instances and Multi-AZ DB clusters with parameter groups. Amazon RDS defines parameter groups with default settings. You can also define your own parameter groups with customized settings.”

CDC for MySQL

Similarly, Debezium has detailed instructions regarding configuring CDC for MySQL. Like PostgreSQL, MySQL requires a custom DB parameter group.

CDC for SQL Server

Lastly, Debezium has detailed instructions for configuring CDC with Microsoft SQL Server. Enabling CDC requires enabling CDC on the SQL Server database and table(s) and creating a new filegroup and associated file. Debezium recommends locating change tables in a different filegroup than you use for source tables. CDC is only supported with Microsoft SQL Server Standard Edition and higher; Express and Web Editions are not supported.

	— enable the database for CDC
	EXEC msdb.dbo.rds_cdc_enable_db 'tickit'
	USE tickit
	GO
	SELECT * FROM sys.filegroups;
	SELECT * FROM sys.database_files;
	— add new filegroup to database
	ALTER DATABASE tickit ADD FILEGROUP CDC_FG1;
	— add new file to filegroup
	ALTER DATABASE tickit
	ADD FILE
	(
	NAME = cdc_data1,
	FILENAME = 'D:\rdsdbdata\DATA\sqldba_data1.ndf',
	SIZE = 500 MB,
	FILEGROWTH = 50 MB
	) TO FILEGROUP CDC_FG1;
	— enabling cdc for a sql server table
	EXEC sys.sp_cdc_enable_table
	@source_schema = N'crm',
	@source_name = N'user',
	@role_name = N'admin',
	@filegroup_name = N'CDC_FG1',
	@supports_net_changes = 0
	GO

view raw ms_sql_cdc.sql hosted with ❤ by GitHub

Kafka Connect Source Connectors

There is a Kafka Connect source connector for each of the three Amazon RDS databases, all of which use Debezium. CDC is performed, moving changes from the three databases into separate Kafka topics in Apache Avro format using the source connectors. The connector configuration is nearly identical whether you are using Amazon MSK Connect or a self-managed version of Kafka Connect.

As shown above, I am using the UI for Apache Kafka by Provectus, self-managed on Amazon EKS, in this post.

PostgreSQL Source Connector

The source_connector_postgres_kafka_avro_tickit source connector captures all changes to the three tables in the Amazon RDS for PostgreSQL ticket database’s ems schema: category, event, and venue. These changes are written to three corresponding Kafka topics as Avro-format messages: tickit.ems.category, tickit.ems.event, and tickit.ems.venue. The messages are transformed by the connector using Debezium’s unwrap transform. Schemas for the messages are written to Apicurio Registry.

	{
	"name": "source_connector_postgres_kafka_avro_tickit",
	"config": {
	"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
	"database.dbname": "tickit",
	"database.hostname": "${secretManager:demo/postgres:host}",
	"database.password": "${secretManager:demo/postgres:password}",
	"database.port": 5432,
	"database.server.name": "tickit",
	"database.user": "${secretManager:demo/postgres:username}",
	"decimal.handling.mode": "double",
	"key.converter": "io.apicurio.registry.utils.converter.AvroConverter",
	"key.converter.apicurio.registry.auto-register": "true",
	"key.converter.apicurio.registry.find-latest": "true",
	"key.converter.apicurio.registry.url": "http://localhost:8080/apis/registry/v2",
	"plugin.name": "pgoutput",
	"schema.name.adjustment.mode": "avro",
	"table.include.list": "ems.category,ems.event,ems.venue",
	"tasks.max": 1,
	"topic.prefix": "tickit",
	"transforms": "unwrap",
	"transforms.unwrap.add.fields": "name,op,db,table,schema,lsn,source.ts_ms",
	"transforms.unwrap.delete.handling.mode": "rewrite",
	"transforms.unwrap.drop.tombstones": "true",
	"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
	"value.converter": "io.apicurio.registry.utils.converter.AvroConverter",
	"value.converter.apicurio.registry.auto-register": "true",
	"value.converter.apicurio.registry.find-latest": "true",
	"value.converter.apicurio.registry.url": "http://localhost:8080/apis/registry/v2"
	}
	}

view raw source_connector_postgres_kafka_avro_tickit.properties hosted with ❤ by GitHub

MySQL Source Connector

The source_connector_mysql_kafka_avro_tickit source connector captures all changes to the three tables in the Amazon RDS for PostgreSQL ecomm database: date, listing, and sale. These changes are written to three corresponding Kafka topics as Avro-format messages: tickit.ecomm.date, tickit.ecomm.listing, and tickit.ecomm.sale.

	{
	"name": "source_connector_mysql_kafka_avro_tickit",
	"config": {
	"connector.class": "io.debezium.connector.mysql.MySqlConnector",
	"database.hostname": "${secretManager:demo/mysql:host}",
	"database.include.list": "ecomm",
	"database.password": "${secretManager:demo/mysql:password}",
	"database.port": 3306,
	"database.server.id": 184054,
	"database.user": "${secretManager:demo/mysql:username}",
	"decimal.handling.mode": "double",
	"include.schema.changes": "true",
	"key.converter": "io.apicurio.registry.utils.converter.AvroConverter",
	"key.converter.apicurio.registry.auto-register": "true",
	"key.converter.apicurio.registry.find-latest": "true",
	"key.converter.apicurio.registry.url": "http://localhost:8080/apis/registry/v2",
	"schema.name.adjustment.mode": "avro",
	"schema.history.internal.kafka.bootstrap.servers": "${secretManager:demo/mysql:bootstrap.servers}",
	"schema.history.internal.kafka.topic": "schemahistory.tickit",
	"tasks.max": 1,
	"topic.prefix": "tickit",
	"transforms": "unwrap",
	"transforms.unwrap.add.fields": "name,op,db,table,source.ts_ms",
	"transforms.unwrap.delete.handling.mode": "rewrite",
	"transforms.unwrap.drop.tombstones": "true",
	"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
	"value.converter": "io.apicurio.registry.utils.converter.AvroConverter",
	"value.converter.apicurio.registry.auto-register": "true",
	"value.converter.apicurio.registry.find-latest": "true",
	"value.converter.apicurio.registry.url": "http://localhost:8080/apis/registry/v2"
	}
	}

view raw source_connector_mysql_kafka_avro_tickit.properties hosted with ❤ by GitHub

SQL Server Source Connector

Lastly, the source_connector_mssql_kafka_avro_tickit source connector captures all changes to the single user table in the Amazon RDS for SQL Server ticket database’s crm schema. These changes are written to a corresponding Kafka topic as Avro-format messages: tickit.crm.user.

	{
	"name": "source_connector_mssql_kafka_avro_tickit",
	"config": {
	"connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
	"database.encrypt": "false",
	"database.hostname": "${secretManager:demo/mssql:host}",
	"database.names": "tickit",
	"database.password": "${secretManager:demo/mssql:password}",
	"database.port": 1433,
	"database.user": "${secretManager:demo/mssql:username}",
	"decimal.handling.mode": "double",
	"include.schema.changes": "true",
	"key.converter": "io.apicurio.registry.utils.converter.AvroConverter",
	"key.converter.apicurio.registry.auto-register": "true",
	"key.converter.apicurio.registry.find-latest": "true",
	"key.converter.apicurio.registry.url": "http://localhost:8080/apis/registry/v2",
	"schema.name.adjustment.mode": "avro",
	"schema.history.internal.kafka.bootstrap.servers": "${secretManager:demo/mssql:bootstrap.servers}",
	"schema.history.internal.kafka.topic": "schemahistory.tickit",
	"table.include.list": "crm.user",
	"tasks.max": 1,
	"topic.prefix": "tickit",
	"transforms": "unwrap",
	"transforms.unwrap.add.fields": "name,op,db,table,source.ts_ms",
	"transforms.unwrap.delete.handling.mode": "rewrite",
	"transforms.unwrap.drop.tombstones": "true",
	"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
	"value.converter": "io.apicurio.registry.utils.converter.AvroConverter",
	"value.converter.apicurio.registry.auto-register": "true",
	"value.converter.apicurio.registry.find-latest": "true",
	"value.converter.apicurio.registry.url": "http://localhost:8080/apis/registry/v2"
	}
	}

view raw source_connector_mssql_kafka_avro_tickit.properties hosted with ❤ by GitHub

If using a self-managed version of Kafka Connect, we can deploy, manage, and monitor the source and sink connectors using Kafka Connect’s RESTful API (Connect API).

Once the three source connectors are running, we should see seven new Kafka topics corresponding to the seven TICKIT database tables from the three Amazon RDS database sources. There will be approximately 425K messages consuming 147 MB of space in Avro’s binary format.

Avro’s binary format and the use of a separate schema ensure the messages consume minimal Kafka storage space. For example, the average message Value (payload) size in the tickit.ecomm.sale topic is a minuscule 98 Bytes. Resource management is critical when you are dealing with hundreds of millions or often billions of daily Kafka messages as a result of the CDC process.

From within the Apicurio Registry UI, we should now see Value schema artifacts for each of the seven Kafka topics.

Also from within the Apicurio Registry UI, we can examine details of individual Value schema artifacts.

Kafka Connect Sink Connector

In addition to the three source connectors, there is a single Kafka Connect sink connector, sink_connector_kafka_s3_avro_tickit. The sink connector copies messages from the seven Kafka topics, all prefixed with tickit, to the Bronze area of our Amazon S3-based data lake.

The connector uses Confluent’s S3 Sink Connector (S3SinkConnector). Like Kafka, the connector writes all changes to Amazon S3 in Apache Avro format, with the schemas already stored in Apicurio Registry.

	{
	"name": "sink_connector_kafka_s3_avro_tickit",
	"config": {
	"behavior.on.null.values": "ignore",
	"connector.class": "io.confluent.connect.s3.S3SinkConnector",
	"flush.size": 10000,
	"format.class": "io.confluent.connect.s3.format.avro.AvroFormat",
	"key.converter": "io.apicurio.registry.utils.converter.AvroConverter",
	"key.converter.apicurio.registry.auto-register": "true",
	"key.converter.apicurio.registry.find-latest": "true",
	"key.converter.apicurio.registry.url": "http://localhost:8080/apis/registry/v2",
	"locale": "en-US",
	"partitioner.class": "io.confluent.connect.storage.partitioner.DailyPartitioner",
	"rotate.schedule.interval.ms": 60000,
	"s3.bucket.name": "<your_data_lake_s3_bucket>",
	"s3.part.size": 5242880,
	"s3.region": "us-east-1",
	"schema.compatibility": "NONE",
	"schema.generator.class": "io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator",
	"storage.class": "io.confluent.connect.s3.storage.S3Storage",
	"table.name.format": "${topic}",
	"tasks.max": 1,
	"timestamp.extractor": "Wallclock",
	"timezone": "UTC",
	"topics.dir": "cdc_hudi_data_lake/bronze",
	"topics.regex": "tickit.(.*)",
	"value.converter": "io.apicurio.registry.utils.converter.AvroConverter",
	"value.converter.apicurio.registry.auto-register": "true",
	"value.converter.apicurio.registry.find-latest": "true",
	"value.converter.apicurio.registry.url": "http://localhost:8080/apis/registry/v2&quot;
	}
	}

view raw sink_connector_kafka_s3_avro_tickit.properties hosted with ❤ by GitHub

Once the sink connector and the three source connectors are running with Kafka Connect, we should see a series of seven subdirectories within the Bronze area of the data lake.

Confluent offers a choice of data partitioning strategies. The sink connector we have deployed uses the Daily Partitioner. According to the documentation, the io.confluent.connect.storage.partitioner.DailyPartitioner is equivalent to the TimeBasedPartitioner with path.format='year'=YYYY/'month'=MM/'day'=dd and partition.duration.ms=86400000 (one day for one S3 object in each daily directory). Below is an example of Avro files (S3 objects) containing changes to the sale table. The objects are partitioned by the year, month, and day they were written to the S3 bucket.

Message Transformation

Previously, while discussing the Kafka Connect source connectors, I mentioned that the connector transforms the messages using the unwrap transform. By default, the changes picked up by Debezium during the CDC process contain several additional Debezium-related fields of data. An example of an untransformed message generated by Debezium from the PostgreSQL sale table is shown below.

When the PostgreSQL connector first connects to a particular PostgreSQL database, it starts by performing a consistent snapshot of each database schema. All existing records are captured. Unlike an UPDATE ("op" : "u") or a DELETE ("op" : "d"), these initial snapshot messages, like the one shown below, represent a READ operation ("op" : "r"). As a result, there is no before data only after.

	{
	"before" : null,
	"after" : {
	"full.ecomm.sale.Value" : {
	"salesid" : 1,
	"listid" : 1,
	"sellerid" : 36861,
	"buyerid" : 21191,
	"eventid" : 7872,
	"dateid" : 1875,
	"qtysold" : 4,
	"pricepaid" : 728.0,
	"commission" : 109.2,
	"saletime" : "2/18/2020 02:36:48"
	}
	},
	"source" : {
	"version" : "2.1.2.Final",
	"connector" : "mysql",
	"name" : "full",
	"ts_ms" : 1677415227000,
	"snapshot" : {
	"string" : "first_in_data_collection"
	},
	"db" : "ecomm",
	"sequence" : null,
	"table" : {
	"string" : "sale"
	},
	"server_id" : 0,
	"gtid" : null,
	"file" : "mysql-bin-changelog.074478",
	"pos" : 154,
	"row" : 0,
	"thread" : null,
	"query" : null
	},
	"op" : "r",
	"ts_ms" : {
	"long" : 1677415227078
	},
	"transaction" : null
	}

view raw sale_message_full.json hosted with ❤ by GitHub

According to Debezium’s documentation, “Debezium provides a single message transformation [SMT] that crosses the bridge between the complex and simple formats, the UnwrapFromEnvelope SMT.” Below, we see the results of Debezium’s unwrap transform of the same message. The unwrap transform flattens the nested JSON structure of the original message, adds the __deleted field, and includes fields based on the list included in the source connector’s configuration. These fields are prefixed with a double underscore (e.g., __table).

	{
	"salesid" : 1,
	"listid" : 1,
	"sellerid" : 36861,
	"buyerid" : 21191,
	"eventid" : 7872,
	"dateid" : 1875,
	"qtysold" : 4,
	"pricepaid" : 728.0,
	"commission" : 109.2,
	"saletime" : "2/18/2020 02:36:48",
	"__name" : {
	"string" : "tickit"
	},
	"__op" : {
	"string" : "r"
	},
	"__db" : {
	"string" : "ecomm"
	},
	"__table" : {
	"string" : "sale"
	},
	"__source_ts_ms" : {
	"long" : 1677379962000
	},
	"__deleted" : {
	"string" : "false"
	}
	}

view raw sale_message_transformed.json hosted with ❤ by GitHub

In the next section, we examine Apache Hudi will manage the data lake. An example of the same message, managed with Hudi, is shown below. Note the five additional Hudi data fields, prefixed with _hoodie_.

	{
	"_hoodie_commit_time": "20230226135257126",
	"_hoodie_commit_seqno": "20230226135257126_0_69229",
	"_hoodie_record_key": "1",
	"_hoodie_partition_path": "__table=sale",
	"_hoodie_file_name": "773bbbb7-b6ec-4ade-9f08-942726584e42-0_0-35-50_20230226135257126.parquet",
	"salesid": 1,
	"listid": 1,
	"sellerid": 36861,
	"buyerid": 21191,
	"eventid": 7872,
	"dateid": 1875,
	"qtysold": 4,
	"pricepaid": 728,
	"commission": 109.2,
	"saletime": "2/18/2020 02:36:48",
	"__name": "tickit",
	"__op": "r",
	"__db": "ecomm",
	"__table": "sale",
	"__source_ts_ms": 1677418032000,
	"__deleted": "false"
	}

view raw sale_message_hudi.json hosted with ❤ by GitHub

Apache Hudi

With database changes flowing into the Bronze area of our data lake, we are ready to use Apache Hudi to provide data lake capabilities such as ACID transactional guarantees, time travel, rollback, and schema evolution. Using Hudi allows us to store and perform analytics on a specific time-bound version of the data in the data lake. Without Hudi, a query for a single row of data could return multiple results, such as an initial CREATE record, multiple UPDATE records, and a DELETE record.

DeltaStreamer

Using Hudi’s DeltaStreamer, we will continuously stream changes from the Bronze area of the data lake to a Silver area, which Hudi manages. We will run DeltaStreamer continuously, similar to an Apache Spark Structured Streaming job, using Apache Spark on Amazon EMR. Running DeltaStreamer requires using a spark-submit command that references a series of configuration files. There is a common base set of configuration items and a group of configuration items specific to each database table. Below, we see the base configuration, base.properties.

	hoodie.upsert.shuffle.parallelism=2
	hoodie.insert.shuffle.parallelism=2
	hoodie.delete.shuffle.parallelism=2
	hoodie.bulkinsert.shuffle.parallelism=2
	hoodie.datasource.hive_sync.mode=hms
	hoodie.datasource.hive_sync.use_jdbc=false
	hoodie.datasource.hive_sync.assume_date_partitioning=false
	hoodie.datasource.hive_sync.database=tickit_cdc_hudi
	hoodie.datasource.hive_sync.partition_extractor_class=org.apache.hudi.hive.MultiPartKeysValueExtractor
	hoodie.datasource.write.hive_style_partitioning=true
	# 1,024 * 1,024 * 128 = 134,217,728 (128 MB)
	hoodie.parquet.small.file.limit=134217728
	hoodie.index.type=GLOBAL_BLOOM
	hoodie.bloom.index.update.partition.path=true

view raw base.properties hosted with ❤ by GitHub

The base configuration is referenced by each of the table-specific configuration files. For example, below, we see the tickit.ecomm.sale.properties configuration file for DeltaStreamer.

	include=base.properties
	hoodie.datasource.hive_sync.partition_fields=__table
	hoodie.datasource.hive_sync.table=sale
	hoodie.datasource.write.partitionpath.field=__table
	hoodie.datasource.write.recordkey.field=salesid
	hoodie.deltastreamer.schemaprovider.registry.url=http://http://<your_registry_host>:<port>/apis/ccompat/v6/subjects/tickit.ecomm.sale-value/versions/latest
	hoodie.deltastreamer.source.dfs.root=s3://<your_data_lake_s3_bucket>/cdc_hudi_data_lake/bronze/tickit.ecomm.sale/

view raw tickit.ecomm.sale.properties hosted with ❤ by GitHub

To run DeltaStreamer, you can submit an EMR Step or use EMR’s master node to run a spark-submit command on the cluster. Below, we see an example of the DeltaStreamer spark-submit command using the Merge on Read (MoR) Hudi table type.

	DATA_LAKE_BUCKET="<your_data_lake_s3_bucket>"
	TARGET_TABLE="tickit.ecomm.sale"
	spark-submit \
	–name %{TARGET_TABLE} \
	–jars /usr/lib/spark/jars/spark-avro.jar,/usr/lib/hudi/hudi-utilities-bundle.jar \
	–conf spark.sql.catalogImplementation=hive \
	–conf spark.yarn.submit.waitAppCompletion=false \
	–class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer `ls /usr/lib/hudi/hudi-utilities-bundle.jar` \
	–props file://${PWD}/${TARGET_TABLE}.properties \
	–table-type MERGE_ON_READ \
	–source-ordering-field __source_ts_ms \
	–source-class org.apache.hudi.utilities.sources.AvroDFSSource \
	–schemaprovider-class org.apache.hudi.utilities.schema.SchemaRegistryProvider \
	–target-table ${TARGET_TABLE} \
	–target-base-path s3://${DATA_LAKE_BUCKET}/cdc_hudi_data_lake/silver/${TARGET_TABLE}/ \
	–enable-sync \
	–continuous \
	–op UPSERT \
	> ${TARGET_TABLE}.log 2>&1 &

view raw HoodieDeltaStreamerMoR.sh hosted with ❤ by GitHub

Next, we see the same example of the DeltaStreamer spark-submit command using the Copy on Write (CoW) Hudi table type.

	DATA_LAKE_BUCKET="<your_data_lake_s3_bucket>"
	TARGET_TABLE="tickit.ecomm.sale"
	spark-submit \
	–name %{TARGET_TABLE} \
	–jars /usr/lib/spark/jars/spark-avro.jar,/usr/lib/hudi/hudi-utilities-bundle.jar \
	–conf spark.sql.catalogImplementation=hive \
	–conf spark.yarn.submit.waitAppCompletion=false \
	–class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer `ls /usr/lib/hudi/hudi-utilities-bundle.jar` \
	–props file://${PWD}/${TARGET_TABLE}.properties \
	–table-type COPY_ON_WRITE \
	–source-ordering-field __source_ts_ms \
	–source-class org.apache.hudi.utilities.sources.AvroDFSSource \
	–schemaprovider-class org.apache.hudi.utilities.schema.SchemaRegistryProvider \
	–target-table ${TARGET_TABLE} \
	–target-base-path s3://${DATA_LAKE_BUCKET}/cdc_hudi_data_lake/silver/${TARGET_TABLE}/ \
	–enable-sync \
	–continuous \
	–op UPSERT \
	> ${TARGET_TABLE}.log 2>&1 &

view raw HoodieDeltaStreamerCoW.sh hosted with ❤ by GitHub

For this post’s demonstration, we will run a single-table DeltaStreamer Spark job, with one process for each table using CoW. Alternately, Hudi offers HoodieMultiTableDeltaStreamer, a wrapper on top of HoodieDeltaStreamer, which enables the ingestion of multiple tables at a single time into Hudi datasets. Currently, HoodieMultiTableDeltaStreamer only supports sequential processing of tables to be ingested and COPY_ON_WRITE storage type.

Below, we see an example of three DeltaStreamer Spark jobs running continuously for the date, listing, and sale tables.

Once DeltaStreamer is up and running, we should see a series of subdirectories within the Silver area of the data lake, all managed by Apache Hudi.

Within each subdirectory, partitioned by the table name, is a series of Apache Parquet files, along with other Apache Hudi-specific files. The specific folder structure and files depend on MoR or Cow.

AWS Glue Data Catalog

For this post, we are using an AWS Glue Data Catalog, an Apache Hive-compatible metastore, to persist technical metadata stored in the Silver area of the data lake, managed by Apache Hudi. The AWS Glue Data Catalog database, tickit_cdc_hudi, will be automatically created the first time DeltaStreamer runs.

Using DeltaStreamer with the table type of MERGE_ON_READ, there would be two tables in the AWS Glue Data Catalog database for each original table. According to Amazon’s EMR documentation, “Merge on Read (MoR) — Data is stored using a combination of columnar (Parquet) and row-based (Avro) formats. Updates are logged to row-based delta files and are compacted as needed to create new versions of the columnar files.” Hudi creates two tables in the Hive metastore for MoR, a table with the name that you specified, which is a read-optimized view (appended with _ro), and a table with the same name appended with _rt, which is a real-time view. You can query both tables.

According to Amazon’s EMR documentation, “Copy on Write (CoW) — Data is stored in a columnar format (Parquet), and each update creates a new version of files during a write. CoW is the default storage type.” Using COPY_ON_WRITE with DeltaStreamer, there is only a single Hudi table in the AWS Glue Data Catalog database for each corresponding database table.

Examining an individual table in the AWS Glue Data Catalog database, we can see details like the table schema, location of underlying data in S3, input/output formats, Serde serialization library, and table partitions.

Database Changes and Data Lake

We are using Kafka Connect, Apache Hudi, and Hudi’s DeltaStreamer to maintain data parity between our databases and the data lake. Let’s look at an example of how a simple data change is propagated from a database to Kafka, then to the Bronze area of the data lake, and finally to the Hudi-managed Silver area of the data lake, written using the CoW table format.

First, we will make a simple update to a single record in the MySQL database’s sale table with the salesid = 200.

	UPDATE ecomm.sale s
	SET s.buyerid = 11694
	s.qtysold = 3,
	s.pricepaid = 600.00,
	s.commission = 90.00
	WHERE s.salesid = 200;

view raw sale_update.sql hosted with ❤ by GitHub

Almost immediately, we can see the change picked up by the Kafka Connect Debezium MySQL source connector.

Almost immediately, in the Bronze area of the data lake, we can see we have a new Avro-formatted file (S3 object) containing the updated record in the partitioned sale subdirectory.

If we examine the new object in the Bronze area of the data lake, we will see a row representing the updated record with the salesid = 200. Note how the operation is now an UPDATE versus a READ ("op" : "u").

	{
	"salesid" : 200,
	"listid" : 214,
	"sellerid" : 31484,
	"buyerid" : 11694,
	"eventid" : 3272,
	"dateid" : 1891,
	"qtysold" : 4,
	"pricepaid" : 600.0,
	"commission" : 90.0,
	"saletime" : "3/6/2020 02:18:11",
	"__name" : {
	"string" : "tickit"
	},
	"__op" : {
	"string" : "u"
	},
	"__db" : {
	"string" : "ecomm"
	},
	"__table" : {
	"string" : "sale"
	},
	"__source_ts_ms" : {
	"long" : 1677467818000
	},
	"__deleted" : {
	"string" : "false"
	}
	}

view raw sale_message_updated.json hosted with ❤ by GitHub

Next, in the corresponding Silver area of the data lake, managed by Hudi, we should also see a new Parquet file that contains a record of the change. In this example, the file was written approximately 26 seconds after the original change to the database. This is the end-to-end time from database change to when the updated data is queryable in the data lake.

Similarly, if we examine the new object in the Silver area of the data lake, we will see a row representing the updated record with the salesid = 200. The record was committed approximately 15 seconds after the original database change.

	{
	"_hoodie_commit_time": "20230227031713915",
	"_hoodie_commit_seqno": "20230227031713915_0_2168",
	"_hoodie_record_key": "200",
	"_hoodie_partition_path": "__table=sale",
	"_hoodie_file_name": "cce62f4e-a111-4aae-bfd1-2c5af1c6bdeb-0_0-82-84_20230227031713915.parquet",
	"salesid": 200,
	"listid": 214,
	"sellerid": 31484,
	"buyerid": 11694,
	"eventid": 3272,
	"dateid": 1891,
	"qtysold": 4,
	"pricepaid": 600,
	"commission": 90,
	"saletime": "3/6/2020 02:18:11",
	"__name": "tickit",
	"__op": "u",
	"__db": "ecomm",
	"__table": "sale",
	"__source_ts_ms": 1677467818000,
	"__deleted": "false"
	}

view raw sale_message_updated_hudi.json hosted with ❤ by GitHub

Querying Hudi Tables with Amazon EMR

Using an EMR Notebook, we can query the updated database record stored in Amazon S3 using Hudi’s CoW table format. First, running a simple Spark SQL query for the record with salesid = 200, returns the latest record, reflecting the changes as indicated by the UPDATE operation value (u) and the _hoodie_commit_time = 2023-02-27 03:17:13.915 UTC.

Hudi Time Travel Query

We can also run a Hudi Time Travel Query, with an as.of.instance set to some arbitrary point in the future. Again, the latest record is returned, reflecting the changes as indicated by the UPDATE operation value (u) and the _hoodie_commit_time = 2023-02-27 03:17:13.915 UTC.

We can run the same Hudi Time Travel Query with an as.of.instance set to on or after the original records were written by DeltaStreamer (_hoodie_commit_time = 2023-02-27 03:13:50.642), but some arbitrary time before the updated record was written (_hoodie_commit_time = 2023-02-27 03:17:13.915). This time, the original record is returned as indicated by the READ operation value (r) and the _hoodie_commit_time = 2023-02-27 03:13:50.642. This is one of the strengths of Apache Hudi, the ability to query data in the present or at any point in the past.

Hudi Point in Time Query

We can also run a Hudi Point in Time Query, with a time range set from the beginning of time until some arbitrary point in the future. Again, the latest record is returned, reflecting the changes as indicated by the UPDATE operation value (u) and the _hoodie_commit_time = 2023-02-27 03:17:13.915.

We can run the same Hudi Point in Time Query but change the time range to two arbitrary time values, both before the updated record was written (_hoodie_commit_time = 2023-02-26 22:39:07.203). This time, the original record is returned as indicated by the READ operation value (r) and the _hoodie_commit_time of 2023-02-27 03:13:50.642. Again, this is one of the strengths of Apache Hudi, the ability to query data in the present or at any point in the past.

Querying Hudi Tables with Amazon Athena

We can also use Amazon Athena and run a SQL query against the AWS Glue Data Catalog database’s sale table to view the updated database record stored in Amazon S3 using Hudi’s CoW table format. The operation (op) column’s value now indicates an UPDATE (u).

How are Deletes Handled?

In this post’s architecture, deleted database records are signified with an operation ("op") field value of "d" for a DELETE and a __deleted field value of true.

https://itnext.io/media/f7c4ad62dcd45543fdf63834814a5aab

Back to our previous Jupyter notebook example, when rerunning the Spark SQL query, the latest record is returned, reflecting the changes as indicated by the DELETE operation value (d) and the _hoodie_commit_time = 2023-02-27 03:52:13.689.

Alternatively, we could use an additional Spark SQL filter statement to prevent deleted records from being returned (e.g., df.__op != "d").

Since we only did what is referred to as a soft delete in Hudi terminology, we can run a time travel query with an as.of.instance set to some arbitrary time before the deleted record was written (_hoodie_commit_time = 2023-02-27 03:52:13.689). This time, the original record is returned as indicated by the READ operation value (r) and the _hoodie_commit_time = 2023-02-27 03:13:50.642. We could also use a later as.of.instance to return the version of the record reflecting the the UPDATE operation value (u). This also applies to other query types such as point-in-time queries.

Conclusion

In this post, we learned how to build a near real-time transactional data lake on AWS using one possible architecture. The data lake was built using a combination of open source software (OSS) and fully-managed AWS services. Red Hat’s Debezium, Apache Kafka, and Kafka Connect were used for change data capture (CDC). In addition, Apache Spark, Apache Hudi, and Hudi’s DeltaStreamer were used to manage the data lake. To complete our architecture, we used several fully-managed AWS services, including Amazon RDS, Amazon MKS, Amazon EKS, AWS Glue, and Amazon EMR.

This blog represents my own viewpoints and not of my employer, Amazon Web Services (AWS). All product names, logos, and brands are the property of their respective owners.