Apache Hudi – Codeswarm

Apache Hudi (Hadoop Upserts and Incrementals) integrates with Apache Spark, making it possible to perform Change Data Capture (CDC) using Spark and Scala. Hudi enables incremental data processing by tracking changes (inserts, updates, deletes) and efficiently managing these on top of distributed storage like HDFS or cloud stores (e.g., S3, GCS).

In many business scenarios, we need to capture changes (inserts, updates, and deletes) from a source system and apply them incrementally to your data lake. Hudi’s upsert and incremental queries are well-suited for this scenario.

Set up Apache Hudi and Spark dependencies:

libraryDependencies ++= Seq(
  "org.apache.hudi" %% "hudi-spark3-bundle" % "X.Y.Z",
  "org.apache.spark" %% "spark-core" % "3.3.0",
  "org.apache.spark" %% "spark-sql" % "3.3.0"
)

Replace X.Y.Z with the appropriate version of Hudi and ensure Spark is aligned with the same version.

Create Hudi Table in Spark:

To manage updates and inserts in a table (i.e., CDC), Hudi uses two main table types:

Copy on Write (COW): More write-intensive, but gives consistent reads.
Merge on Read (MOR): Faster writes but slightly complex reads due to incremental merges.

You can create a Hudi table using Spark DataFrame API. First, load your data into a DataFrame, and then apply Hudi configurations.

import org.apache.spark.sql.{SaveMode, SparkSession}
val spark = SparkSession.builder
  .appName("Hudi CDC Example")
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .getOrCreate()
// Example data in DataFrame (e.g., CDC data)
val df = spark.read.json("path_to_cdc_data")
// Hudi table write options
val hudiOptions = Map[String, String](
  "hoodie.table.name" -> "hudi_cdc_table",
  "hoodie.datasource.write.operation" -> "upsert",  // To handle CDC events
  "hoodie.datasource.write.recordkey.field" -> "primary_key_column",  // Primary key for CDC
  "hoodie.datasource.write.precombine.field" -> "timestamp_column",   // Helps resolve duplicate rows
  "hoodie.datasource.hive_sync.enable" -> "true",                     // Optional if syncing to Hive
  "hoodie.datasource.hive_sync.database" -> "hudi_db",
  "hoodie.datasource.hive_sync.table" -> "hudi_cdc_table",
  "hoodie.datasource.hive_sync.partition_fields" -> "partition_column",
  "hoodie.datasource.write.hive_style_partitioning" -> "true"
)
// Writing to Hudi table using DataFrame API
df.write.format("hudi")
  .options(hudiOptions)
  .mode(SaveMode.Append)  // Append or Overwrite as needed
  .save("path_to_hudi_table")

Read Incremental Data

After loading data incrementally using Hudi, you can read the changes (new inserts, updates, and deletes) since the last commit.

// Reading Hudi data incrementally from the latest commit
val incrementalOptions = Map(
  "hoodie.datasource.query.type" -> "incremental",
  "hoodie.datasource.read.begin.instanttime" -> "20230101000000" // Specify the commit timestamp to start from
)
val incrementalDF = spark.read.format("hudi")
  .options(incrementalOptions)
  .load("path_to_hudi_table")
incrementalDF.show(false)

Handle Updates and Deletes in CDC

Apache Hudi allows you to manage updates and deletes through the upsert operation. When upserting data, Hudi automatically checks for the primary key and handles whether it should insert, update, or delete the record.

Handling Deletes

For deletions, the key field should be provided, and the operation should be set to delete in Hudi write options.

val deleteOptions = Map[String, String](
  "hoodie.table.name" -> "hudi_cdc_table",
  "hoodie.datasource.write.operation" -> "delete",  // For deletes
  "hoodie.datasource.write.recordkey.field" -> "primary_key_column"
)
val deleteDF = // Your dataframe containing the primary keys to delete
deleteDF.write.format("hudi")
  .options(deleteOptions)
  .mode(SaveMode.Append)
  .save("path_to_hudi_table")

Key Configurations

hoodie.datasource.write.operation: Choose upsert, insert, or delete depending on your CDC operation,
hoodie.datasource.write.recordkey.field: Set this to the primary key column, which is used to identify unique rows,
hoodie.datasource.write.precombine.field: Used for deduplication in the case of multiple updates to the same record,
hoodie.datasource.query.type: Used when reading, with options like snapshot (full read) or incremental (read only changed records).

Advantages of Using Apache Hudi for CDC

Efficient Storage: Hudi manages the data lifecycle by compacting files and removing unnecessary duplicates.
Streaming Use Case: With Hudi, you can efficiently consume CDC streams (e.g., from Kafka) and apply updates.
Incremental Processing: Hudi allows for reading incremental changes, which improves processing efficiency when only new or changed data needs to be processed.

Example with Spark Structured Streaming (CDC Streaming)

If you’re processing CDC events from a stream (like Kafka), you can combine Spark Structured Streaming with Hudi as follows:

val streamDF = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "cdc-topic")
  .load()
// Write stream to Hudi table
streamDF.writeStream
  .format("hudi")
  .options(hudiOptions)
  .outputMode("append")
  .start("path_to_hudi_table")

This allows for continuous ingestion of changes and real-time updates into your Hudi dataset.

Summary

By combining Apache Hudi with Spark in Scala, you can efficiently implement a CDC system that supports upserts, deletes, and real-time data ingestion with minimal overhead. Hudi simplifies the management of large-scale distributed datasets by offering incremental data processing and optimized storage.

Tag: Apache Hudi

Apache Hudi CDC (Change Data Capture) Use case with Scala/Spark