Apache Hudi

YugabyteDB CDC

This integration allows continuous and incremental data ingestion from YugabyteDB into analytical processes, leveraging the power of Apache Hudi.

Set up Kafka and schema registry

1. Download docker-compose.yaml from the CDC-examples folder and run all the specified containers.

   This installs the Confluent schema registry, Control Center, ZooKeeper, Kafka, YugabyteDB Debezium Kafka Connector, Grafana, and Prometheus containers. Configure the ports as required.

   This example uses port 8091 for the schema registry as follows:

   schema-registry:
     image: confluentinc/cp-schema-registry:7.2.1
     hostname: schema-registry
     container_name: schema-registry
     depends_on:
       - broker
     ports:
       - "8091:8091"
     environment:
       SCHEMA_REGISTRY_HOST_NAME: schema-registry
       SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:29092'
       SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8091
   

2. Run the docker containers:

   docker-compose up -d
   

Set up and configure gRPC CDC stream ID in YugabyteDB

Create a database stream ID for a specific database (for example, demo):

1. Assign your current database node address to the "{IP}" variable as follows:

   export IP=10.23.16.6
   

2. Run the following yb-admin command from the YugabyteDB node bin directory or /home/yugabyte/tserver/bin directory of any database node of your YugabyteDB cluster:

   ./yb-admin --master_addresses ${IP}:7100 create_change_data_stream ysql.demo implicit all
   

   This outputs a stream ID, such as a4f8291c3737419dbe4feee5a1b19aee.

Deploy the Kafka source connector

Run the Kafka connector with the following command:

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{
  "name": "cdc-demo",
  "config": {
    "connector.class": "io.debezium.connector.yugabytedb.YugabyteDBConnector",
    "database.hostname": "'$IP'",
    "database.port": "5433",
    "tasks.max": "3",
    "database.master.addresses": "'$IP':7100",
    "database.user": "yugabyte",
    "database.password": "xxxxxxx",
    "database.dbname": "demo",
    "database.server.name": "dbs",
    "table.include.list": "public.cdctest",
    "database.streamid": "a4f8291c3737419dbe4feee5a1b19aee",
    "transforms":"pgcompatible",
    "transforms.pgcompatible.type":"io.debezium.connector.yugabytedb.transforms.PGCompatible",
    "key.converter.schemas.enable": "true",
    "value.converter.schemas.enable": "true",
    "tombstones.on.delete":"false",
    "key.converter":"io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url":"http://schema-registry:8091",
    "value.converter":"io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url":"http://schema-registry:8091"
  }
}'

Validate the schema and verify Kafka topics

Launch the Confluent Control Center and verify the schema details and messages in the relevant topics.

Access the Control Center at http://<docker_container_IP_or_VM>:9021.

Start populating the data from YugabyteDB, and ensure you are able to see the messages in Control Center –> Topics –> Messages (for example, http://172.18.0.2:9021/clusters/management/topics/cdc.public.cdctest/message-viewer).

Install Apache Hudi

1. Build Apache Hudi from source. Modify the source files to account for differences between PostgreSQL and YugabyteDB CDC emissions as follows:

   • In DebeziumConstants.java, comment out or remove lines related to xmin.

   • In PostgresDebeziumSource.java, comment out or remove parameters related to xmin.

2. Run Maven to build Hudi as follows:

   mvn clean package -DskipTests
   

Run a Spark job using HoodieDeltaStreamer

1. Create a spark-config.properties file as follows:

   spark.serializer=org.apache.spark.serializer.KryoSerializer
   spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog
   spark.sql.hive.convertMetastoreParquet=false
   

2. Run the Spark job as follows:

   spark-submit --class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer \
     --packages org.apache.hudi:hudi-spark3.4-bundle_2.12:0.14.0 \
     --master local[*] \
     "/your_folder/hudi-release-0.14.0/packaging/hudi-utilities-bundle/target/hudi-utilities-bundle_2.12-0.14.0.jar" \
     --table-type MERGE_ON_READ \
     --source-class org.apache.hudi.utilities.sources.debezium.PostgresDebeziumSource \
     --payload-class org.apache.hudi.common.model.debezium.PostgresDebeziumAvroPayload \
     # Adjust the target base path for the Hudi table as per your setup
     --target-base-path file:///tmp/hoodie/dbs-cdctest \
     # Adjust the Hudi table name as per your setup
     --target-table dbs_cdctest \
     --source-ordering-field _event_origin_ts_ms \
     --continuous \
     --source-limit 4000000 \
     --min-sync-interval-seconds 20 \
     --hoodie-conf bootstrap.servers=localhost:9092 \
     --hoodie-conf schema.registry.url=http://localhost:8091 \
     --hoodie-conf hoodie.deltastreamer.schemaprovider.registry.url=http://localhost:8091/subjects/dbs.public.cdctest-value/versions/latest \
     --hoodie-conf hoodie.deltastreamer.source.kafka.value.deserializer.class=io.confluent.kafka.serializers.    KafkaAvroDeserializer \
     --hoodie-conf hoodie.deltastreamer.source.kafka.topic=dbs.public.cdctest \
     --hoodie-conf auto.offset.reset=earliest \
     # Adjust the recordkey file as per your setup
     --hoodie-conf hoodie.datasource.write.recordkey.field=sno \
     # Adjust the partition name as per your setup
     --hoodie-conf hoodie.datasource.write.partitionpath.field=name \
     --hoodie-conf hoodie.datasource.write.keygenerator.class=org.apache.hudi.keygen.SimpleKeyGenerator \
     # Adjust the path for the Spark configuration file as per your setup
     --props /your_folder/spark-config.properties
   

   Adjust paths and filenames as per your environment setup.

Query the Hudi table

1. Verify the Hudi table is created in /tmp/hoodie/dbs-cdctest as follows:

   cd /tmp/hoodie/dbs-cdctest
   ls -atlr
   

2. Start a Spark shell to query the Hudi table as follows:

   spark-shell
   

3. Import necessary libraries and read data from the Hudi table:

   import org.apache.hudi.DataSourceReadOptions._
   import org.apache.hudi.config.HoodieWriteConfig._
   import org.apache.spark.sql.SaveMode._
   import spark.implicits._
   
   val basePath = "file:///tmp/hoodie/dbs-cdctest"
   val cdcDF = spark.read.format("hudi").load(basePath)
   cdcDF.createOrReplaceTempView("cdcdemo")
   
   spark.sql("SELECT _hoodie_commit_time, sno, name, _change_operation_type FROM cdcdemo").show()
   

Verify

Perform database operations on the cdctest table in YugabyteDB and verify that the changes are reflected in the Hudi table by running queries through the Spark shell.

1. Insert and update records in the cdctest table in your YugabyteDB database as follows:

   -- Example of insert statement
   INSERT INTO public.cdctest (sno, name) VALUES (826, 'Test User 1');
   
   -- Example of update statement
   UPDATE public.cdctest SET name = 'Updated Test User 1' WHERE sno = 826;
   

2. Start a new Spark shell with spark-shell and import the necessary classes and read the data from the Hudi table:

   import org.apache.hudi.DataSourceReadOptions._
   import org.apache.hudi.config.HoodieWriteConfig._
   import org.apache.spark.sql.SaveMode._
   import spark.implicits._
   
   val basePath = "file:///tmp/hoodie/dbs-cdctest"
   val cdcDF = spark.read.format("hudi").load(basePath)
   cdcDF.createOrReplaceTempView("cdcdemo")
   
   spark.sql("SELECT _hoodie_commit_time, sno, name, _change_operation_type FROM cdcdemo").show()
   

   You can see the changes reflected from the cdctest table in YugabyteDB through the CDC connector into the Hudi table. The _change_operation_type field indicates whether the operation was an update (u), insert (c), or delete (d).

3. Delete a record from the cdctest table in YugabyteDB:

   DELETE FROM public.cdctest WHERE sno = 826;
   

4. Verify deletion in Hudi Table with the following query in a Spark shell:

   spark.sql("SELECT _hoodie_commit_time, sno, name, _change_operation_type FROM cdcdemo").show()
   

   The record with sno = 826 is deleted from the Hudi table due to the propagated delete event.