hudi-agent commented on code in PR #19597: URL: https://github.com/apache/hudi/pull/19597#discussion_r3767334117
########## website/docs/hoodie_streaming_ingestion.md: ########## @@ -628,6 +628,77 @@ Using `org.apache.hudi.utilities.sources.SqlFileBasedSource` allows setting the table. SQL file path should be configured using this hoodie config: `hoodie.streamer.source.sql.file = 'hdfs://xxx/source.sql'` +#### Debezium + +Hudi Streamer can keep a Hudi table in sync with an upstream database by ingesting change data capture (CDC) events +produced by [Debezium](https://debezium.io/). Debezium publishes each change as an Avro message on a Kafka topic and +registers the schema with a Confluent schema registry. The Debezium sources read that topic, flatten the nested Debezium +change envelope into ordinary table columns, and apply the resulting inserts, updates and deletes to the target table. + +There is one source and one matching payload class per database: + +| Database | Source class | Payload class | +|------------|---------------------------------------------------------------------|---------------------------------------------------------------------| +| PostgreSQL | `org.apache.hudi.utilities.sources.debezium.PostgresDebeziumSource` | `org.apache.hudi.common.model.debezium.PostgresDebeziumAvroPayload` | +| MySQL | `org.apache.hudi.utilities.sources.debezium.MysqlDebeziumSource` | `org.apache.hudi.common.model.debezium.MySqlDebeziumAvroPayload` | + +Note that the two halves spell MySQL differently: the source is `Mysql...` while the payload is `MySql...`. + +Both sources read Avro and require a schema registry, so set `--schemaprovider-class` to +`org.apache.hudi.utilities.schema.SchemaRegistryProvider` and point `hoodie.streamer.schemaprovider.registry.url` at the +subject for the topic. The Kafka value deserializer already defaults to +`io.confluent.kafka.serializers.KafkaAvroDeserializer`, so `hoodie.streamer.source.kafka.value.deserializer.class` only +needs setting in order to override it. + +A property file for a PostgreSQL table: + +```properties +hoodie.streamer.source.kafka.topic=postgres.public.customers +hoodie.streamer.schemaprovider.registry.url=http://localhost:8081/subjects/postgres.public.customers-value/versions/latest +bootstrap.servers=localhost:9092 +auto.offset.reset=earliest + +hoodie.datasource.write.recordkey.field=id +``` + +and the job that reads it: + +```java +[hoodie]$ spark-submit \ + --packages org.apache.hudi:hudi-utilities-slim-bundle_2.12:1.2.0,org.apache.hudi:hudi-spark3.5-bundle_2.12:1.2.0 \ + --class org.apache.hudi.utilities.streamer.HoodieStreamer `ls packaging/hudi-utilities-slim-bundle/target/hudi-utilities-slim-bundle-*.jar` \ + --props file://${PWD}/debezium-source.properties \ + --schemaprovider-class org.apache.hudi.utilities.schema.SchemaRegistryProvider \ + --source-class org.apache.hudi.utilities.sources.debezium.PostgresDebeziumSource \ + --payload-class org.apache.hudi.common.model.debezium.PostgresDebeziumAvroPayload \ + --source-ordering-field _event_lsn \ + --target-base-path file:///tmp/hudi-debezium-customers \ + --target-table customers \ + --table-type MERGE_ON_READ \ + --op UPSERT \ + --continuous +``` + +The record key must be the primary key of the upstream table, so that later changes to a row update it in place. A +Merge-on-Read table suits the small, frequent writes a CDC stream produces, but Copy-on-Write works as well. + +**Ordering.** Change events can reach Kafka out of order, so the payload decides which version of a row wins rather than +relying on arrival order. For PostgreSQL that is the log sequence number in `_event_lsn`, which the payload reads Review Comment: 🤖 The prose explains that MySQL orders on the derived `_event_seq` column, but the only end-to-end example uses PostgreSQL with `--source-ordering-field _event_lsn`. A MySQL user has to infer that they must swap in the `Mysql...` source, the `MySql...` payload, and `--source-ordering-field _event_seq`. It might help to state explicitly that for MySQL you set `--source-ordering-field _event_seq` (alongside the MySQL source/payload classes), so the ordering configuration is unambiguous for both databases. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
