基于debezium(mongo)-kafka-starrocks的cdc实践
先说结论
解决了mongodb能支持cdc的before消息体,但因为 StarRocks Connector for Kafka
惊天巨bug,本轮实践验证失败。
目标
网上的资料真是少,怎么找都没有找到能支持mongodb删除的debezium cdc的中文资料,看了半天英文资料,实践一下,这里主要验证一下 StarRocks Connector for Kafka
source
: MongoDbConnectorsink
: StarRocksSinkConnector
我们的目标是将mongodb表数据实时CDC进入starrocks数仓,支持删除操作。
前置设施
mongodb
首先我们得确保mongodb是6.0,因为6.0后支持了changeStreamPreAndPostImages变更流。
原理:在6.0以前,因为mongodb不支持changeStreamPreAndPostImages
变更流,无法在cdc语句显示出before,所以当进行删除操作的时候,按照debizium的格式规范,无法完成删除操作
如果需要在debezium cdc输出的语句上显示before信息需要
- 开启mongodb版本 6.0 中的新增功能
changeStreamPreAndPostImages
。1
2
3
4
5
6
7use user
db.runCommand({
collMod: "user",
changeStreamPreAndPostImages: {
enabled: true
}
}) - 在debezium mongodb connector的
capture.mode
上使用以下任意一个change_streams_with_pre_image
输出before,不输出afterchange_streams_update_full_with_pre_image
输出before和after模式 描述 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_update_full的基础上增加,增加before字段,用于输出变化前变化后的数据的内容 1
2
3
4
5
6
7
8
9
10{
"name": "source-user-mongodb-connector",
"config": {
"connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
"mongodb.connection.string": "mongodb://xxx:xxx@172.16.0.11:27017,172.16.0.15:27017/?&authSource=admin",
"collection.include.list": "mydb.user",
"topic.prefix": "user-cdc",
"capture.mode": "change_streams_update_full_with_pre_image"
}
}
kafka
开启了自动创建topic。
如果没有自动开启创建,需要为debezium手工创建
config.storage.topic
offset.storage.topic
status.storage.topic
手工创建,参考Kafka Connect 接入 CKafka,有许多debezium的要求,所以建议自动创建。
部署 debezium/connect
使用容器分布式部署方式,使得支持通过rest api管理connector,
通过volumes,将starrocks-kafka-connector也集成支持进去
这里需要注意debezium/connect镜像已经将常见的mysql、mongodb、sqlserver集成了进去了
1 | version: '2' |
流入:mongo->kafka ✅
mongo user 表 流入 kafka
user表先插入一条数据吧,方便提交任务的时候,先走全量快照
1 | db.getCollection("user").insert( { |
可通过 http://ip:8083/connectors
,提交source mongodb connector任务
1 | curl --location 'http://ip:8083/connectors' \ |
我们来看看增删改查对应的消息结构
- 创建
创建的时候 before是null,after是创建后的
json字符串
,op是c1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26{
"schema": {...},
"payload": {
"before": null,
"after": "{\"_id\": {\"$oid\": \"6676dccb988e8964430f7e46\"},\"name\": \"junyao\"}",
"updateDescription": null,
"source": {
"version": "2.4.2.Final",
"connector": "mongodb",
"name": "user",
"ts_ms": 1719065804000,
"snapshot": "false",
"db": "mydb",
"sequence": null,
"rs": "cmgo-6t0e5t7x_0",
"collection": "user",
"ord": 1,
"lsid": null,
"txnNumber": null,
"wallTime": 1719065804058
},
"op": "c",
"ts_ms": 1719065804054,
"transaction": null
}
} - 更新
before是更新前,after是更新后,op是u
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30{
"schema": {...},
"payload": {
"before": "{\"_id\": {\"$oid\": \"6676dbe5988e8964430f7e45\"},\"name\": \"tony\"}",
"after": "{\"_id\": {\"$oid\": \"6676dbe5988e8964430f7e45\"},\"name\": \"danny\"}",
"updateDescription": {
"removedFields": null,
"updatedFields": "{\"name\": \"danny\"}",
"truncatedArrays": null
},
"source": {
"version": "2.4.2.Final",
"connector": "mongodb",
"name": "user",
"ts_ms": 1719065650000,
"snapshot": "false",
"db": "mydb",
"sequence": null,
"rs": "cmgo-6t0e5t7x_0",
"collection": "user",
"ord": 1,
"lsid": null,
"txnNumber": null,
"wallTime": 1719065650298
},
"op": "u",
"ts_ms": 1719065650298,
"transaction": null
}
} - 删除
before是删除前,after是删除后
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26{
"schema": {...},
"payload": {
"before": "{\"_id\": {\"$oid\": \"6676dbe5988e8964430f7e45\"},\"name\": \"danny\"}",
"after": null,
"updateDescription": null,
"source": {
"version": "2.4.2.Final",
"connector": "mongodb",
"name": "user",
"ts_ms": 1719065732000,
"snapshot": "false",
"db": "mydb",
"sequence": null,
"rs": "cmgo-6t0e5t7x_0",
"collection": "user",
"ord": 1,
"lsid": null,
"txnNumber": null,
"wallTime": 1719065732402
},
"op": "d",
"ts_ms": 1719065732399,
"transaction": null
}
}
流出:kafka -> StarRocksSinkConnector ⚠️
kafka流出starrocks,目前未走通
- 创建starrocks表
1
2
3
4
5
6
7
8CREATE TABLE IF NOT EXISTS `scm`.`ods_test_user` (
`_id` STRING ,
`name` STRING NULL
) ENGINE=OLAP
PRIMARY KEY (_id) DISTRIBUTED BY HASH (_id) PROPERTIES
(
"replication_num" = "1"
) - 提交任务
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24curl --location 'http://ip:8083/connectors' \
--header 'Content-Type: application/json' \
--data '{
"name": "sink-test23-user-cdc",
"config": {
"connector.class": "com.starrocks.connector.kafka.StarRocksSinkConnector",
"topics": "user-cdc23.test.user",
"key.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable": "true",
"value.converter.schemas.enable": "false",
"starrocks.http.url": "xxx.xxx.xxx.xxx:38030",
"starrocks.topic2table.map": "user-cdc23.test.user:ods_test_user",
"starrocks.username": "root",
"starrocks.password": "xxx",
"starrocks.database.name": "scm",
"sink.properties.strip_outer_array": "true",
"transforms": "addfield,unwrap",
"transforms.addfield.type": "com.starrocks.connector.kafka.transforms.AddOpFieldForDebeziumRecord",
"transforms.unwrap.type": "io.debezium.connector.mongodb.transforms.ExtractNewDocumentState",
"transforms.unwrap.drop.tombstones": "true",
"transforms.unwrap.delete.handling.mode": "rewrite"
}
}'
这里需要注意,按照starrocks的文档导入 Debezium CDC 格式数据,对于mongodb的transforms.unwrap.type存在差异,因为我是mongodb6.0,所以我看了一下最新的 MongoDB New Document State Extraction,相比较 New Record State Extraction,有以下的差异
- New Record State Extraction
1
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
- MongoDB New Document State Extraction
1
"transforms.unwrap.type": "io.debezium.connector.mongodb.transforms.ExtractNewDocumentState",
需要调整成 io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
调整好提交任务,看报错日志。
1 | root@starrockscluster-cn-1:/opt/starrocks# curl http://starrockscluster-cn-1.starrockscluster-cn-search.starrocks.svc.cluster.local:8040/api/_load_error_log?file=error_log_864bbf459bb530a6_44aa9dc5433d4c0 |
失败,差评!
流出:kafka -> StarRocksSinkConnector 失败后换个思路 ⚠️
换个思路,我可不可以在 mongo source流入kafka的时候就处理好格式? 于是我改了下,添加以下内容到source
1 | "transforms": "addfield,unwrap", |
- mongo source得到了这样的数据结构
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23curl --location 'ip:8083/connectors' \
--header 'Accept: application/json' \
--header 'Content-Type: application/json' \
--data-raw '{
"name": "source-test26-user-cdc",
"config": {
"connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
"mongodb.connection.string": "mongodb://mongouser:XXX@172.16.0.11:27017,172.16.0.15:27017/?&authSource=admin",
"collection.include.list": "test.user",
"topic.prefix": "user-cdc26",
"capture.mode": "change_streams_update_full_with_pre_image",
"snapshot.mode": "initial",
"key.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable": "true",
"value.converter.schemas.enable": "false",
"transforms": "addfield,unwrap",
"transforms.addfield.type": "com.starrocks.connector.kafka.transforms.AddOpFieldForDebeziumRecord",
"transforms.unwrap.type": "io.debezium.connector.mongodb.transforms.ExtractNewDocumentState",
"transforms.unwrap.drop.tombstones": "true",
"transforms.unwrap.delete.handling.mode": "rewrite"
}
}'1
2
3
4
5{
"_id": "66793958d767977a330e9245",
"name": "jy",
"__deleted": false
} - starrocks sink
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19curl --location 'http://ip:8083/connectors' \
--header 'Content-Type: application/json' \
--data '{
"name": "sink-test27-user-cdc",
"config": {
"connector.class": "com.starrocks.connector.kafka.StarRocksSinkConnector",
"topics": "user-cdc27.test.user",
"key.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable": "true",
"value.converter.schemas.enable": "false",
"starrocks.http.url": "159.75.190.XXX:38030",
"starrocks.topic2table.map": "user-cdc27.test.user:ods_test_user",
"starrocks.username": "root",
"starrocks.password": "XXX",
"starrocks.database.name": "scm",
"sink.properties.strip_outer_array": "true"
}
}' - 验证
我们来验证一下增删改查一下
- 创建⚠️成功同步数据到starrocks,并没有添加__op:0, 没有添加的时候系统默认是__op:0
1
2
3
4
5{
"_id": "66793e78d767977a330e9246",
"name": "ivan",
"__deleted": false
} - 更新⚠️成功同步数据到starrocks,并没有添加__op:0, 没有添加的时候系统默认是__op:0
1
2
3
4
5{
"_id": "66793958d767977a330e9245",
"name": "danny",
"__deleted": false
} - 删除⚠️失败,starrocks并没有删除,并没有添加__op:1, 没有添加的时候系统默认是__op:0,所以无法删除,应该是做了更新操作
1
2
3
4
5{
"_id": "66793e78d767977a330e9246",
"name": "ivan",
"__deleted": true
}
- 创建
我们可以看到实际transforms出来的并没有包含__op,
而按照StarRocks 的主键表的UPSERT 和 DELETE 操作
我们发现,消息体内没有__op,都会按照UPSERT,这就能理解为什么没有成功删除了。
差评!