deepakpanda93 commented on code in PR #19597: URL: https://github.com/apache/hudi/pull/19597#discussion_r3773187387
########## 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: Good catch — fixed in 0d7625ac. I did verify the suggested value before applying it, because at first it looked like it might be wrong. `HoodieTableConfig` reconciles `MySqlDebeziumAvroPayload` to ordering fields `_event_bin_file,_event_pos`, and the class itself carries a matching constant: ```java public static final String ORDERING_FIELDS = FLATTENED_FILE_COL_NAME + "," + FLATTENED_POS_COL_NAME; ``` so the two-column form, not `_event_seq`, is what Hudi's own machinery uses. Worth chasing down before publishing a recommendation. `_event_seq` turns out to be right, for a reason that makes the whole thing clearer. `AbstractDebeziumAvroPayload.preCombine` decides a duplicate by comparing `orderingVal` — the *configured* ordering field — not a column read off the record: ```java if (((Comparable) oldValue.getOrderingValue()).compareTo(orderingVal) > 0) { ``` So the ordering field genuinely matters for both databases; it isn't decorative. And at 1.2.0 `MySqlDebeziumAvroPayload` overrides `preCombine` with `isCurrentSeqLatest`, which does: ```java String[] currentFilePos = currentSeq.split("\\."); long currentFileNum = Long.parseLong(currentFilePos[0]); long currentPos = Long.parseLong(currentFilePos[1]); ``` That wants exactly the `file.pos` string — which is what `MysqlDebeziumSource` derives into `_event_seq` (`fileId.substring(fileId.lastIndexOf('.') + 1).concat("." + pos)`). Passing `_event_bin_file,_event_pos` there would not split into two numeric halves. The two-column form is what reconciliation fills in by itself; `_event_seq` is what a user passes. The derivation is present in all five 1.x tags, so this holds across the whole version range the PR touches. The reason each database's mechanism is worth distinguishing: Postgres reads `_event_lsn` off the record in `shouldPickCurrentRecord` *and* uses it as the ordering value, whereas MySQL reads `_event_seq` off the record in `shouldPickCurrentRecord` but compares the ordering value in `preCombine`. Both paths converge on the same column per database, which is why one flag per database is the correct guidance. The change adds the MySQL variant of the arguments right after the example: ```java --source-class org.apache.hudi.utilities.sources.debezium.MysqlDebeziumSource \ --payload-class org.apache.hudi.common.model.debezium.MySqlDebeziumAvroPayload \ --source-ordering-field _event_seq \ ``` and the ordering paragraph now says the column has to be passed as `--source-ordering-field`, with the reason, instead of only describing where the value comes from. Build still passes with the warning block byte-identical to the baseline at `d11a5b0adee4`; the MySQL source, payload and `_event_seq` render on `next` and all five 1.x copies. One note on the review itself: the summary says the class names, ordering fields, delete marker, deserializer default and every-batch override behaviour were all verified against the tagged source. Those do check out. But the merge-mode point in the PR description is the one I'd still like a committer to weigh in on — whether to leave it out, as here, or document it with an explicit version qualifier, given `PAYLOADS_UNDER_DEPRECATION` exists only at 1.1.1 and 1.2.0. -- 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]
