前置
服务器上需要安装jdk11
jdk下载地址
kafka安装
官网下载地址
安装教程
debezium__9">debezium 安装
运行 Debezium 连接器需要 Java 11 或更高版本
Debezium 并不是一个独立的软件，而是很多个 Kafka 连接器的总称。这些 Kafka 连接器分别对应不同的数据库，比如 MySQL、Oracle 等。按 Kafka 连接器的常见命名规则，可能我们会把它们叫做 MySQL Kafka Source Connector 之类。
部署
debezium_15">1.下载对应版本的debezium插件
插件地址
2.文件解压
将下载的文件解压,将解压后的文件放到kafka的plugin文件夹下(该plugin文件夹为自己创建的plugin文件夹)*,例如
3. 通过 kafka connect部署
kafka connect有两种部署方式，一是单机部署，二是分布式部署。单机部署配置kafka/config/connect-standalone.properties 文件，分布式部署则配置kafka/config/connect-distributed.properties。分布式部署支持通过rest api管理connector
此处是单机部署，配置文件为kafka/config/connect-standalone.properties，主要修改以下内容：
# plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
plugin.path=/usr/local/kafka/plugin
4.启动kafka-connect
需要先启动kafka
bin/connect-standalone.sh config/connect-standalone.properties
debezium_39">5.创建对应的debezium配置文件
curl -X POST http://${debezium所在服务器}:8083/connectors
{"name": "cdc-connector","config": {"connector.class": "io.debezium.connector.mongodb.MongoDbConnector","mongodb.connection.string": "mongodb://root:123456@192.168.2.18:27017,192.168.2.19:27017/?authSource=admin","collection.include.list": "db_cdc_1.c_cdc_2","topic.prefix": "mycdc","capture.mode":"change_streams"}
}
- 如果需要在cdc输出的语句上显示before信息,需要开启mongodb版本 6.0 中的新增功能changeStreamPreAndPostImages,并且在capture.mode上使用change_streams_with_pre_image或change_streams_update_full_with_pre_image
- 如果capture.mode未设置成change_streams_with_pre_image或change_streams_update_full_with_pre_image的话,在进行删除时cdc输出会没有before信息
db.runCommand({collMod: "对应的controllerName", changeStreamPreAndPostImages: {enabled: true}
})
例如:
use db_cdc_1
db.runCommand({collMod: "c_cdc_2", changeStreamPreAndPostImages: {enabled: true}
}){"name": "cdc-connector","config": {"connector.class": "io.debezium.connector.mongodb.MongoDbConnector","mongodb.connection.string": "mongodb://root:123456@192.168.2.18:27017,192.168.2.19:27017/?authSource=admin","collection.include.list": "db_cdc_1.c_cdc_2","topic.prefix": "mycdc","capture.mode":"change_streams_with_pre_image"}
}
重点参数
| 参数 | 描述 |
|---|---|
| connector.class | 固定值io.debezium.connector.mongodb.MongoDbConnector |
| mongodb.connection.string | mongodb连接信息 |
| collection.include.list | 需要监听的具体collection |
| topic.prefix | kafkaTopic前缀 |
| capture.mode | 输出模式(默认:change_streams_update_full) |
capture.mode
| 模式 | 描述 |
|---|---|
| change_streams | 输出变化流,但是在进行update操作时,不输出after字段 |
| change_streams_update_full | 在change_streams的基础上,增加after字段,用于输出现在变化后的数据的内容 |
| change_streams_with_pre_image | 在change_streams的基础上,增加before字段的输出,但需要进行配置 |
| change_streams_update_full_with_pre_image | 在change_streams_with_pre_image的基础上增加,增加after字段,用于输出现在变化后的数据的内容 |
其他未使用参数
| 参数 | 描述 |
|---|---|
| database.include.list | 需要监听的具体database |
| database.exclude.list | 不监听的database(不要与database.include.list填写相同的db) |
| collection.exclude.list | 不监听的collection(不要与collection.include.list填写相同的collection) |
| snapshot.mode | 指定在连接器启动时执行快照的条件。Initial(默认:重头开始)当连接器启动时，如果没有在其偏移主题中检测值，它会执行数据库的快照。never(从当前位置开始)当连接器启动时，它会跳过快照进程，并立即开始将数据库记录的操作流传输到 oplog。 |
更多参数请参考
cdc结果
原数据
{"userId": "1000000","allPoints": 190,"createTime": {"$date": "2024-04-25T13:31:59.678Z"},"updateTime": {"$date": "2024-04-25T13:31:59.678Z"}
}
添加数据
capture.mode两种模式输出结果一样
push数据
{$push: {"history":{"historyId": "1","changerPoints": 0,"beforePoints": 0,"afterPoints": 0,"status": "0","createTime": {"$date": "2024-01-01T16:00:00.000Z"},"comment": "测试数据","versionNo": 0}},
}
第一次添加
{"payload": {"before": null,"after": null,"updateDescription": {"removedFields": null,"updatedFields": "{\"history\": [{\"historyId\": \"1\", \"changerPoints\": 0, \"beforePoints\": 0, \"afterPoints\": 0, \"status\": \"0\", \"createTime\": {\"$date\": \"2024-01-01T16:00:00.000Z\"}, \"comment\": \"测试数据\", \"versionNo\": 0}]}","truncatedArrays": null}}
第二次添加
{"payload": {"before": null,"after": null,"updateDescription": {"removedFields": null,"updatedFields": "{\"history.1\": {\"historyId\": \"2\", \"changerPoints\": 0, \"beforePoints\": 0, \"afterPoints\": 0, \"status\": \"0\", \"createTime\": {\"$date\": \"2024-01-01T16:00:00.000Z\"}, \"comment\": \"测试数据\", \"versionNo\": 0}}","truncatedArrays": null}}
如果capture.mode为change_streams_update_full,则会在after字段上显示现在修改的这条数据的完整数据,例如
修改数组中的值
{"payload": {"before": null,"after": null,"updateDescription": {"removedFields": null,"updatedFields": "{\"history.1.historyId\": \"100\"}","truncatedArrays": null}}
如果capture.mode为change_streams_update_full,则会在after字段上显示现在修改的这条数据现有的完整数据,例如
pull操作
{$pull: {history: {historyId: "100",},},
}
此时会把现有的所有数据都返回
{"payload": {"before": null,"after": null,"updateDescription": {"removedFields": null,"updatedFields": "{\"history\": [{\"historyId\": \"1\", \"changerPoints\": 0, \"beforePoints\": 0, \"afterPoints\": 0, \"status\": \"0\", \"createTime\": {\"$date\": \"2024-01-01T16:00:00.000Z\"}, \"comment\": \"测试数据\", \"versionNo\": 0}, {\"historyId\": \"2\", \"changerPoints\": 0, \"beforePoints\": 0, \"afterPoints\": 0, \"status\": \"0\", \"createTime\": {\"$date\": \"2024-01-01T16:00:00.000Z\"}, \"comment\": \"测试数据\", \"versionNo\": 0}, {\"historyId\": \"3\", \"changerPoints\": 0, \"beforePoints\": 0, \"afterPoints\": 0, \"status\": \"0\", \"createTime\": {\"$date\": \"2024-01-01T16:00:00.000Z\"}, \"comment\": \"测试数据\", \"versionNo\": 0}, {\"historyId\": \"4\", \"changerPoints\": 0, \"beforePoints\": 0, \"afterPoints\": 0, \"status\": \"0\", \"createTime\": {\"$date\": \"2024-01-01T16:00:00.000Z\"}, \"comment\": \"测试数据\", \"versionNo\": 0}]}","truncatedArrays": null}}
如果capture.mode为change_streams_update_full,则会在after字段上显示现在修改的这条数据现有的完整数据,例如
删除字段
{ "payload": {"before": null,"after": null,"updateDescription": {"removedFields": ["updateTime"],"updatedFields": "{}","truncatedArrays": null}}
如果capture.mode为change_streams_update_full,则会在after字段上显示现在修改的这条数据现有的完整数据(此处删除的是另外一个字段),例如
删除数据
如果capture.mode未设置成change_streams_with_pre_image或change_streams_update_full_with_pre_image的话,在进行删除时cdc输出会没有before信息
通过开启mongodb版本 6.0 中的新增功能changeStreamPreAndPostImages,并且在capture.mode上使用change_streams_with_pre_image或change_streams_update_full_with_pre_image即可解决
db.runCommand({collMod: "对应的controllerName", changeStreamPreAndPostImages: {enabled: true}
})
例如:
use db_cdc_1
db.runCommand({collMod: "c_cdc_2", changeStreamPreAndPostImages: {enabled: true}
}){"name": "cdc-connector","config": {"connector.class": "io.debezium.connector.mongodb.MongoDbConnector","mongodb.connection.string": "mongodb://root:123456@192.168.2.18:27017,192.168.2.19:27017/?authSource=admin","collection.include.list": "db_cdc_1.c_cdc_2","topic.prefix": "mycdc","capture.mode":"change_streams_with_pre_image"}
}
