Db2 到 Kafka
| 功能 | 说明 |
|---|---|
Schema Migration | If the specified Topic after mapping does not exist in the Target, BladePipe will automatically create the Topic, allowing setting the number of partitions. |
Full Data Migration | Migrate data by sequentially scanning data in tables and writing it in batches to the message-oriented middleware. |
Incremental Data Sync | Sync of common DML like INSERT, UPDATE, DELETE is supported. |
Subscription Modification | Add, delete, or modify the subscribed tables with support for historical data migration. For more information, see Modify Subscription. |
Position Resetting | Reset positions by LSN to consume the incremental data logs from a specific LSN. |
Metadata Retrieval | Retrieve the target metadata with filtering conditions or target primary keys set from the source table. |
高级功能
| 功能 | 说明 |
|---|---|
消息格式 | 支持以下消息格式,文档:消息格式说明
|
Topic Mapping Rules | By default, Topic is formed by connecting source instance id, database, and table with . in between (e.g., my-vgpq6q097174t6t.dingtax.app_key). Also, it supports the mapping rules, namely, keeping the name the same as that in Source, converting the text to lowercase, converting the text to uppercase. |
Table-level Topic | Create Topics corresponding to the tables in the Source, and the table partitions can be obtained automatically. |
DDL Dedicated Topic | Allow specifying a Topic for DDL. If not specified, DDL time is placed in partition 0 of the Topic created from the corresponding table. |
Scheduled Full Data Migration | For more information, see Create Scheduled Full Data DataJob. |
Custom Code | For more information, see Custom Code Processing, Debug Custom Code and Logging in Custom Code. |
Data Filtering Conditions | Support data filtering using WHERE conditions, with SQL-92 as the SQL language. For more information, see Data Filtering. |
使用示例
| 标题 | 详情 |
|---|---|
跨互联网数据互通 (Kafka) | |
Kafka 数据中转校验 | 文档:Kafka 数据中转校验 |
任务参数
| 参数名称 | 说明 |
|---|---|
fullBatchSize | 全量写入对端单批数据条数,对迁移性能影响大 |
fullPagingCount | 关系型数据库源端扫描分页大小 |
scanParallel | 源端扫描并发数,如果数据源类型为关系型数据库,则为并发扫描表数量 |
eventStoreSize | 内存队列大小(可能为事件数量或者内存大小(byte)) |
maxTxsPerIteration | Worker 扫描的事务数量 |
iterateIntervalMs | 事务扫描的间隔毫秒数 |
snapshotRead | 读源端数据库的方式是否采用全表扫描(select column_name from table) |
filterDDL | 增量是否进行 DDL 同步,值为 true 则不进行 DDL 同步 |
metaFreshCoolMs | 元信息的刷新等待时间 |
metaFreshCount | 元信息刷新次数 |
cdcAutoClear | 是否开启 CDC 表自动清理 |
cdcClearIntervalMin | 清理 CDC 表的周期(分钟) |
dbHeartBeatEnable | 配置对源端数据库是否开启心跳 |
Tips: 通用参数配置请参考 通用参数及功能
任务参数
| 参数名称 | 说明 |
|---|---|
schemaFormat | 消息格式,文档:消息格式说明 |
batchWriteSize | 单条消息最大数据条数,超过则拆分消息 |
defaultTopic | 无法找到对应 Topic 的消息则发送到此 Topic (如新增表) |
ddlTopic | 专门发送 DDL 的 Topic, 为空则发送到对应 Topic 的第 0 个分区 |
compressionType | Kafka compression.type 参数, 设置压缩算法, 支持 GZIP, SNAPPY, LZ4, ZSTD 算法 |
batchSize | Kafka batch.size 参数 |
acks | Kafka acks 参数, 默认 all |
maxRequestBytes | Kafka max.request.size 参数 |
lingerMs | Kafka linger.ms 参数, 默认 1 |
envelopSchemaInclude | 当 schemaFormat 设置为 DEBEZIUM_ENVELOP_JSON_FOR_MQ 时,消息体是否包含 schema 信息 |
customClientProps | 自定义传入到 Kafka Client 参数,JSON 格式,key为参数名,value为参数值。此配置项以最高优先级生效。例如:AWS IAM 访问控制 |
Tips: 通用参数配置请参考 通用参数及功能