{"id":123,"date":"2024-08-31T09:45:00","date_gmt":"2024-08-31T07:45:00","guid":{"rendered":"https:\/\/codeswarm.io\/?p=123"},"modified":"2026-07-19T13:44:47","modified_gmt":"2026-07-19T11:44:47","slug":"apache-hudi-cdc-change-data-capture-use-case-with-scala-spark","status":"publish","type":"post","link":"https:\/\/codeswarm.io\/?p=123","title":{"rendered":"Apache Hudi CDC (Change Data Capture) Use case with Scala\/Spark"},"content":{"rendered":"\n<p class=\"wp-block-paragraph\">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).<\/p>\n\n\n\n<p class=\"wp-block-paragraph\">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\u2019s <strong>upsert<\/strong> and <strong>incremental queries<\/strong> are well-suited for this scenario.<\/p>\n\n\n\n<p class=\"wp-block-paragraph\"><strong>Set up Apache Hudi and Spark dependencies:<\/strong><\/p>\n\n\n\n<pre class=\"EnlighterJSRAW\" data-enlighter-language=\"scala\" data-enlighter-theme=\"\" data-enlighter-highlight=\"\" data-enlighter-linenumbers=\"\" data-enlighter-lineoffset=\"\" data-enlighter-title=\"\" data-enlighter-group=\"\">libraryDependencies ++= Seq(\n  \"org.apache.hudi\" %% \"hudi-spark3-bundle\" % \"X.Y.Z\",\n  \"org.apache.spark\" %% \"spark-core\" % \"3.3.0\",\n  \"org.apache.spark\" %% \"spark-sql\" % \"3.3.0\"\n)<\/pre>\n\n\n\n<p class=\"wp-block-paragraph\">Replace <code>X.Y.Z<\/code> with the appropriate version of Hudi and ensure Spark is aligned with the same version.<\/p>\n\n\n\n<p class=\"wp-block-paragraph\"><strong>Create Hudi Table in Spark<\/strong>:<\/p>\n\n\n\n<p class=\"wp-block-paragraph\">To manage updates and inserts in a table (i.e., CDC), Hudi uses two main table types:<\/p>\n\n\n\n<ul class=\"wp-block-list\">\n<li><strong>Copy on Write (COW)<\/strong>: More write-intensive, but gives consistent reads.<\/li>\n\n\n\n<li><strong>Merge on Read (MOR)<\/strong>: Faster writes but slightly complex reads due to incremental merges.<\/li>\n<\/ul>\n\n\n\n<p class=\"wp-block-paragraph\">You can create a Hudi table using Spark DataFrame API. First, load your data into a DataFrame, and then apply Hudi configurations.<\/p>\n\n\n\n<pre class=\"EnlighterJSRAW\" data-enlighter-language=\"scala\" data-enlighter-theme=\"\" data-enlighter-highlight=\"\" data-enlighter-linenumbers=\"\" data-enlighter-lineoffset=\"\" data-enlighter-title=\"\" data-enlighter-group=\"\">import org.apache.spark.sql.{SaveMode, SparkSession}\nval spark = SparkSession.builder\n  .appName(\"Hudi CDC Example\")\n  .config(\"spark.serializer\", \"org.apache.spark.serializer.KryoSerializer\")\n  .getOrCreate()\n\/\/ Example data in DataFrame (e.g., CDC data)\nval df = spark.read.json(\"path_to_cdc_data\")\n\/\/ Hudi table write options\nval hudiOptions = Map[String, String](\n  \"hoodie.table.name\" -> \"hudi_cdc_table\",\n  \"hoodie.datasource.write.operation\" -> \"upsert\",  \/\/ To handle CDC events\n  \"hoodie.datasource.write.recordkey.field\" -> \"primary_key_column\",  \/\/ Primary key for CDC\n  \"hoodie.datasource.write.precombine.field\" -> \"timestamp_column\",   \/\/ Helps resolve duplicate rows\n  \"hoodie.datasource.hive_sync.enable\" -> \"true\",                     \/\/ Optional if syncing to Hive\n  \"hoodie.datasource.hive_sync.database\" -> \"hudi_db\",\n  \"hoodie.datasource.hive_sync.table\" -> \"hudi_cdc_table\",\n  \"hoodie.datasource.hive_sync.partition_fields\" -> \"partition_column\",\n  \"hoodie.datasource.write.hive_style_partitioning\" -> \"true\"\n)\n\/\/ Writing to Hudi table using DataFrame API\ndf.write.format(\"hudi\")\n  .options(hudiOptions)\n  .mode(SaveMode.Append)  \/\/ Append or Overwrite as needed\n  .save(\"path_to_hudi_table\")<\/pre>\n\n\n\n<p class=\"wp-block-paragraph\"><strong>Read Incremental Data<\/strong><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">After loading data incrementally using Hudi, you can read the changes (new inserts, updates, and deletes) since the last commit.<\/p>\n\n\n\n<pre class=\"EnlighterJSRAW\" data-enlighter-language=\"scala\" data-enlighter-theme=\"\" data-enlighter-highlight=\"\" data-enlighter-linenumbers=\"\" data-enlighter-lineoffset=\"\" data-enlighter-title=\"\" data-enlighter-group=\"\">\/\/ Reading Hudi data incrementally from the latest commit\nval incrementalOptions = Map(\n  \"hoodie.datasource.query.type\" -> \"incremental\",\n  \"hoodie.datasource.read.begin.instanttime\" -> \"20230101000000\" \/\/ Specify the commit timestamp to start from\n)\nval incrementalDF = spark.read.format(\"hudi\")\n  .options(incrementalOptions)\n  .load(\"path_to_hudi_table\")\nincrementalDF.show(false)<\/pre>\n\n\n\n<p class=\"wp-block-paragraph\"><strong>Handle Updates and Deletes in CDC<\/strong><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">Apache Hudi allows you to manage updates and deletes through the <code>upsert<\/code> operation. When upserting data, Hudi automatically checks for the primary key and handles whether it should insert, update, or delete the record.<\/p>\n\n\n\n<p class=\"wp-block-paragraph\"><strong>Handling Deletes<\/strong><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">For deletions, the key field should be provided, and the <code>operation<\/code> should be set to <code>delete<\/code> in Hudi write options.<\/p>\n\n\n\n<pre class=\"EnlighterJSRAW\" data-enlighter-language=\"scala\" data-enlighter-theme=\"\" data-enlighter-highlight=\"\" data-enlighter-linenumbers=\"\" data-enlighter-lineoffset=\"\" data-enlighter-title=\"\" data-enlighter-group=\"\">val deleteOptions = Map[String, String](\n  \"hoodie.table.name\" -> \"hudi_cdc_table\",\n  \"hoodie.datasource.write.operation\" -> \"delete\",  \/\/ For deletes\n  \"hoodie.datasource.write.recordkey.field\" -> \"primary_key_column\"\n)\nval deleteDF = \/\/ Your dataframe containing the primary keys to delete\ndeleteDF.write.format(\"hudi\")\n  .options(deleteOptions)\n  .mode(SaveMode.Append)\n  .save(\"path_to_hudi_table\")<\/pre>\n\n\n\n<p class=\"wp-block-paragraph\"><strong>Key Configurations<\/strong><\/p>\n\n\n\n<ul class=\"wp-block-list\">\n<li><strong><code>hoodie.datasource.write.operation<\/code><\/strong>: Choose <code>upsert<\/code>, <code>insert<\/code>, or <code>delete<\/code> depending on your CDC operation,<\/li>\n\n\n\n<li><strong><code>hoodie.datasource.write.recordkey.field<\/code><\/strong>: Set this to the primary key column, which is used to identify unique rows,<\/li>\n\n\n\n<li><strong><code>hoodie.datasource.write.precombine.field<\/code><\/strong>: Used for deduplication in the case of multiple updates to the same record,<\/li>\n\n\n\n<li><strong><code>hoodie.datasource.query.type<\/code><\/strong>: Used when reading, with options like <code>snapshot<\/code> (full read) or <code>incremental<\/code> (read only changed records).<\/li>\n<\/ul>\n\n\n\n<p class=\"wp-block-paragraph\"><strong>Advantages of Using Apache Hudi for CDC<\/strong><\/p>\n\n\n\n<ol class=\"wp-block-list\">\n<li><strong>Efficient Storage<\/strong>: Hudi manages the data lifecycle by compacting files and removing unnecessary duplicates.<\/li>\n\n\n\n<li><strong>Streaming Use Case<\/strong>: With Hudi, you can efficiently consume CDC streams (e.g., from Kafka) and apply updates.<\/li>\n\n\n\n<li><strong>Incremental Processing<\/strong>: Hudi allows for reading incremental changes, which improves processing efficiency when only new or changed data needs to be processed.<\/li>\n<\/ol>\n\n\n\n<p class=\"wp-block-paragraph\"><strong>Example with Spark Structured Streaming (CDC Streaming)<\/strong><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">If you\u2019re processing CDC events from a stream (like Kafka), you can combine Spark Structured Streaming with Hudi as follows:<\/p>\n\n\n\n<pre class=\"EnlighterJSRAW\" data-enlighter-language=\"scala\" data-enlighter-theme=\"\" data-enlighter-highlight=\"\" data-enlighter-linenumbers=\"\" data-enlighter-lineoffset=\"\" data-enlighter-title=\"\" data-enlighter-group=\"\">val streamDF = spark\n  .readStream\n  .format(\"kafka\")\n  .option(\"kafka.bootstrap.servers\", \"localhost:9092\")\n  .option(\"subscribe\", \"cdc-topic\")\n  .load()\n\/\/ Write stream to Hudi table\nstreamDF.writeStream\n  .format(\"hudi\")\n  .options(hudiOptions)\n  .outputMode(\"append\")\n  .start(\"path_to_hudi_table\")<\/pre>\n\n\n\n<p class=\"wp-block-paragraph\">This allows for continuous ingestion of changes and real-time updates into your Hudi dataset.<\/p>\n\n\n\n<p class=\"wp-block-paragraph\"><strong>Summary<\/strong><\/p>\n\n\n\n<p class=\"wp-block-paragraph\">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.<\/p>\n","protected":false},"excerpt":{"rendered":"<p>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, &hellip; <a href=\"https:\/\/codeswarm.io\/?p=123\" class=\"more-link\">Continue reading <span class=\"screen-reader-text\">Apache Hudi CDC (Change Data Capture) Use case with Scala\/Spark<\/span> <span class=\"meta-nav\">&rarr;<\/span><\/a><\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[24,23],"tags":[20,22,19],"class_list":["post-123","post","type-post","status-publish","format-standard","hentry","category-big-data","category-programming","tag-hudi","tag-spark","tag-scala"],"_links":{"self":[{"href":"https:\/\/codeswarm.io\/index.php?rest_route=\/wp\/v2\/posts\/123","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/codeswarm.io\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/codeswarm.io\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/codeswarm.io\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/codeswarm.io\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=123"}],"version-history":[{"count":6,"href":"https:\/\/codeswarm.io\/index.php?rest_route=\/wp\/v2\/posts\/123\/revisions"}],"predecessor-version":[{"id":129,"href":"https:\/\/codeswarm.io\/index.php?rest_route=\/wp\/v2\/posts\/123\/revisions\/129"}],"wp:attachment":[{"href":"https:\/\/codeswarm.io\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=123"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/codeswarm.io\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=123"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/codeswarm.io\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=123"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}