debezium 是一个方便的cdc connector 可以帮助我们解决好多数据实时变更处理、数据分析、微服务的数据通信
从上次跑简单demo到现在，这个工具是有好多的变更，添加了好多方便的功能，支持了越来越多的数据库。
demo 使用了官方提供的docker-compose 文件
环境准备
- docker-compose 文件
version: '2'
services:
zookeeper:
image: debezium/zookeeper:0.9
ports:
- 2181:2181
- 2888:2888
- 3888:3888
kafka:
image: debezium/kafka:0.9
ports:
- 9092:9092
links:
- zookeeper
environment:
- ZOOKEEPER_CONNECT=zookeeper:2181
sqlserver:
image: microsoft/mssql-server-linux:2017-CU8
ports:
- 1433:1433
environment:
- ACCEPT_EULA=Y
- MSSQL_PID=Standard
- SA_PASSWORD=Password!
connect:
image: debezium/connect:0.9
ports:
- 8083:8083
links:
- kafka
- sqlserver
environment:
- BOOTSTRAP_SERVERS=kafka:9092
- GROUP_ID=1
- CONFIG_STORAGE_TOPIC=my_connect_configs
- OFFSET_STORAGE_TOPIC=my_connect_offsets
- 启动
docker-compose up -d
- 加载sql server 数据
可以使用ui (azure data studio )工具或者命令行（inside docker）,demo 提供了一个sql 脚本文件
cat debezium-sqlserver-init/inventory.sql | docker exec -i tutorial_sqlserver_1 bash -c '/opt/mssql-tools/bin/sqlcmd -U sa -P $SA_PASSWORD'
效果
- 启动connector
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d @register-sqlserver.json
register-sqlserver.json 内容
{"name": "inventory-connector",
"config": {"connector.class" : "io.debezium.connector.sqlserver.SqlServerConnector",
"tasks.max" : "1",
"database.server.name" : "server1",
"database.hostname" : "sqlserver",
"database.port" : "1433",
"database.user" : "sa",
"database.password" : "Password!",
"database.dbname" : "testDB",
"database.history.kafka.bootstrap.servers" : "kafka:9092",
"database.history.kafka.topic": "schema-changes.inventory"
}
}
效果
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d @register-sqlserver.json
HTTP/1.1 201 Created
Date: Tue, 25 Dec 2018 11:26:05 GMT
Location: http://localhost:8083/connectors/inventory-connector
Content-Type: application/json
Content-Length: 462
Server: Jetty(9.4.12.v20180830)
{"name":"inventory-connector","config":{"connector.class":"io.debezium.connector.sqlserver.SqlServerConnector","tasks.max":"1","database.server.name":"server1","database.hostname":"sqlserver","database.port":"1433","database.user":"sa","database.password":"Password!","database.dbname":"testDB","database.history.kafka.bootstrap.servers":"kafka:9092","database.history.kafka.topic":"schema-changes.inventory","name":"inventory-connector"},"tasks":[],"type":null}%测试
- 打开消费者（kafka 命令行）
docker-compose -f docker-compose-sqlserver.yaml exec kafka /kafka/bin/kafka-console-consumer.sh \
--bootstrap-server kafka:9092 \
--from-beginning \
--property print.key=true \
--topic server1.dbo.customers
- 修改数据&&查看效果
说明
注意debezium 本地，0.9 支持sql server
参考资料
https://github.com/debezium/debezium-examples/tree/master/tutorial#using-sql-server
https://debezium.io/