# Debezium Lake Consumer Replicates database CDC events to a lake database ## Lake Consumer Lake debezium server consumer replicates debezium CDC events to destination lake tables. It is possible to replicate source database one to one or run it with `append` mode or `upsert` mode to keep all change events in lake table. When event and key schema enabled (`debezium.format.value.schemas.enable=true`, `debezium.format.key.schemas.enable=true`) destination lake tables created automatically with initial job. #### Configuration properties | Config | Default | Description | |----------------------------------------------------------|-------------------|-----------------------------------------------------------------------------------------------------------------------------------------| | `debezium.sink.lake.database.url` | `` | Lake database jdbc url, example: jdbc:lake://localhost:8000. | | `debezium.sink.lake.database.username` | `` | Lake database user name. | | `debezium.sink.lake.database.password` | `` | Lake database user password. | | `debezium.sink.lake.table-prefix` | `` | Prefix added to destination table names. | | `debezium.sink.lake.upsert` | `true` | Running upsert mode overwriting updated rows. explained below. | | `debezium.sink.lake.upsert-keep-deletes` | `true` | With upsert mode, keeps deleted rows in target table. | | `debezium.sink.lake.destination-regexp` | `` | Regexp to modify destination table. With this its possible to map `table_ptt1`,`table_ptt2` to `table_combined`. | | `debezium.sink.lake.destination-regexp-replace` | `` | Regexp Replace part to modify destination table | | `debezium.sink.batch.batch-size-wait` | `NoBatchSizeWait` | Batch size wait strategy to optimize data files and upload interval. explained below. | | `debezium.sink.lake.database.param.{jdbc.prop.name}` | | Additional jdbc connection config for destination database, example: ?ssl=true here use debezium.sink.lake.database.param.ssl=true. | | `debezium.sink.lake.database.primaryKey` | `` | Lake Table priamryKey for upsert mode, if upsert is true, user must set this config | ### Upsert mode By default, Debezium Lake consumer is running with upsert mode `debezium.sink.lake.upsert=true`. Upsert mode must set Primary Key `debezium.sink.lake.database.primaryKey` and does upsert on target table. For the tables without Primary Key consumer falls back to append mode. > NOTE: If in upsert mode, `lake-debezium-server` support two kind of delete mode: > 1. `hard delete`: In hard delete mode, add `debezium.transforms.unwrap.delete.handling.mode=none`, `debezium.transforms.unwrap.drop.tombstones=false` in config file. Because Debezium generates a tombstone record for each DELETE operation. The default behavior is that ExtractNewRecordState removes tombstone records from the stream. To keep tombstone records in the stream, specify drop.tombstones=false. > 2. `soft delete`: In soft delete mode, user must add the `__deleted` field in target lake table, and add `debezium.transforms.unwrap.delete.handling.mode=rewrite`, `debezium.transforms.unwrap.drop.tombstones=true` in config file. So now we softly delete by make `__deleted` as true. ### Append mode Setting `debezium.sink.lake.upsert=false` will set the operation mode to append. With append mode data deduplication is not done and all received records are appended to destination table. > Note: If user does not set primary key config the operation mode will fall back to append even configuration is set to upsert mode #### Keeping Deleted Records By default `debezium.sink.lake.upsert-keep-deletes=true` keeps deletes in the Lake table, setting it to false will do soft delete from the destination Lake table by making `__deleted` field as true. #### debezium.transforms.unwrap.add.fields If user has this config `debezium.transforms.unwrap.add.fields=op,table,source.ts_ms,db`, make sure your target table has `__op`, `__table`,`__source.ts_ms` and `__db` field or the target table was created by debezium server. ### Optimizing batch size Debezium extracts database events in real time and this could cause too frequent commits which is not optimal for batch processing especially when near realtime data feed is sufficient. To avoid this problem following batch-size-wait classes are used. Batch size wait adds delay between consumer calls to increase total number of events received per call and meanwhile events are collected in memory. This setting should be configured together with `debezium.source.max.queue.size` and `debezium.source.max.batch.size` debezium properties #### NoBatchSizeWait This is default configuration by default consumer will not use any wait. All the events are consumed immediately. #### MaxBatchSizeWait MaxBatchSizeWait uses debezium metrics to optimize batch size, this strategy is more precise compared to DynamicBatchSizeWait. MaxBatchSizeWait periodically reads streaming queue current size and waits until it reaches to `max.batch.size`. Maximum wait and check intervals are controlled by `debezium.sink.batch.batch-size-wait.max-wait-ms` , `debezium.sink.batch.batch-size-wait.wait-interval-ms` properties. example setup to receive ~2048 events per commit. maximum wait is set to 30 seconds, streaming queue current size checked every 5 seconds ```properties debezium.sink.batch.batch-size-wait=MaxBatchSizeWait debezium.sink.batch.metrics.snapshot-mbean=debezium.postgres:type=connector-metrics,context=snapshot,server=debezium debezium.sink.batch.metrics.streaming-mbean=debezium.postgres:type=connector-metrics,context=streaming,server=debezium debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector debezium.source.max.batch.size=2048; debezium.source.max.queue.size=16000"; debezium.sink.batch.batch-size-wait.max-wait-ms=30000 debezium.sink.batch.batch-size-wait.wait-interval-ms=5000 ``` ### include databases and include tables User can use `debezium.source.database.include.list=databaseName1,databaseName2` to monitor the source databases and use `debezium.source.table.include.list=databaseName.tableName1,databaseName.tableName2` to monitor the source tables. ## Debezium Event Flattening Debezium Lake consumer requires event flattening. ```properties debezium.transforms=unwrap debezium.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState # debezium.transforms.unwrap.add.fields=op,table,source.ts_ms,db make sure the target table has there field debezium.transforms.unwrap.add.headers=db debezium.transforms.unwrap.delete.handling.mode=rewrite debezium.transforms.unwrap.drop.tombstones=true ``` ### Lake JDBC Configuring All the properties starting with `debezium.sink.lake.database.param.key` are passed to Lake Jdbc connection. ```properties debezium.sink.lake.database.param.key=value ``` ### Table Name Mapping Target Lake tables are named by following rule : `table-prefix``database.server.name`_`database`_`table` For example: ```properties database.server.name=lake debezium.sink.lake.table-prefix=cdc_ ``` With above config database table = `database.table` is replicated to `lake_cdc_database_table` Or if user don't like this kind of table name mapping, just use `debezium.sink.lake.database.tableName` to point a specific name. [Detailed documentation](https://github.com/lakecloud/lake-jdbc/blob/main/docs/Connection.md) about how to use connection parameters in a Lake jdbc connection. ### Special type convert #### Decimal types Debezium connectors handle decimals according to the setting of the decimal.handling.mode connector configuration property. Specifies how the connector should handle values for DECIMAL and NUMERIC columns: ```properties # precise (the default) represents them precisely using java.math.BigDecimal values represented in change events in a binary form. decimal.handling.mode=precise ``` ```properties # string encodes values as formatted strings, which is easy to consume but semantic information about the real type is lost. decimal.handling.mode=string ``` ```properties # double converts values to approximate double-precision floating-point values. decimal.handling.mode=double ``` #### DateTime types Convert timestamps between different formats such as Unix epoch, strings, and Connect Date/Timestamp types. Applies to individual fields or to the entire value. Use the concrete transformation type designed for the record key (`org.apache.kafka.connect.transforms.TimestampConverter$Key`) or value (`org.apache.kafka.connect.transforms.TimestampConverter$Value`). ##### Examples This configuration snippet shows how to use `TimestampConverter` to transform a Unix epoch (represented as an int64 value) into a formatted date string. ```properties debezium.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState debezium.transforms.a.type=org.apache.kafka.connect.transforms.TimestampConverter$Value debezium.transforms.a.target.type=string debezium.transforms.a.field=a debezium.transforms.a.format=yyyy-MM-dd hh:mm:ss ``` **NOTE:** If your table has `timestamp` or `datetime` type, you need to create your target table in lake manually. Because debezium will convert `timestamp` or `datetime` type to `string` type. ### Example Configuration Read [application.properties.example](../lake-debezium-server-sink/src/main/resources/conf/application.properties.example) ### Problems and Points for improvement - Due to the integration of Debezium Engine, full-load reading phase does not support checkpoint. After a failure, it needs to be re-read. - Currently, only single concurrency is supported, and horizontal scaling is not supported. - When ensuring data consistency, Debezium needs to apply locks to the read databases or tables. Global locks may cause the database to hang, while table-level locks can restrict table reads. So perhaps in the next step, we can draw inspiration from Netflix DBLog's lock-free algorithm to improve these issues. ### Related docs [debezium-mysql-connector](https://debezium.io/documentation/reference/2.0/connectors/mysql.html#mysql-property-table-include-list)