Debezium使用指南

姜同学

发布于 2022-10-27 17:22:27

2.7K0

发布于 2022-10-27 17:22:27

文章被收录于专栏：姜同学姜同学

实时数仓的第一步便是变更数据捕获（CDC）,Debezium就是一款功能非常强大的CDC工具。Debezium是构建于Kafka之上的，将捕获的数据实时的采集到Kafka上

Debezium监控MySQL

监控MySQL的前提是MySQL需要开启binlog日志哦

MySQL开启binlog

MySQL查看是否开启Binlog

show global variables like 'binlog_format';
show global variables like 'log_bin';

下载MySQL connector

wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/1.7.1.Final/debezium-connector-mysql-1.7.1.Final-plugin.tar.gz

mkdir /opt/debezium/
tar -zxvf debezium-connector-mysql-1.7.1.Final-plugin.tar.gz -C /opt/debezium/

修改Kafka配置文件connect-distributed.properties

注意我这里用的kafka为2.12-2.4.1，不同版本的kafka配置可能有所不同配置文件内容如下

# kafka地址，多个地址用英文,隔开
bootstrap.servers=192.168.1.197:9092
group.id=connect-mysql
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
status.storage.topic=connect-mysql-status
status.storage.replication.factor=
offset.flush.interval.ms=
offset.storage.topic=connect-offsets
offset.storage.replication.factor=
config.storage.topic=connect-configs
config.storage.replication.factor=
plugin.path=/opt/debezium/debezium-connector-mysql

启动kafka connector

启动之前记得把debezium MySQL connector里面的jar包拷贝到kafka的libs目录下

/opt/module/kafka-2.4.1/bin/connect-distributed.sh -daemon /opt/module/kafka-2.4.1/config/connect-distributed.properties

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" 192.168.1.197:8083/connectors/ -d @mysql-cdc.json

注册MySQL 连接器

注册连接器的方式也比较简单，kafka连接器发送post请求将配置信息放到请求体就可以了。配置信息如下：

{
    "name": "pins-mysql-connector",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "database.hostname": "192.168.1.95",
        "database.port": "3306",
        "database.user": "ffcdc",
        "database.password": "ucAuzatWUtHkiYyi",
        "database.server.id": "9527",
        "database.server.name": "pins-mysql-cdc",
        "database.include.list": "parkinsondb",
        "database.history.kafka.bootstrap.servers": "192.168.1.197:9092",
        "database.history.kafka.topic": "schema-changes.inventory",
        "table.include.list":"parkinsondb.sys_log"
    }
}

database.include.list配置监控的数据库，不填写的话就监控所有数据库。

Debezium Oracle Connector 的快照模式 snapshot.mode

snapshot.mode 支持的参数配置，这个参数只在连接器在第一次启动时起作用

参数值	描述
initial(默认)	连接器执行数据库的初始一致性快照，快照完成后，连接器开始为后续数据库更改流式传输事件记录。
initial_only	连接器只执行数据库的初始一致性快照，不允许捕获任何后续更改的事件。
schema_only	连接器只捕获所有相关表的表结构，不捕获初始数据，但是会同步后续数据库的更改记录。
schema_only_recovery	设置此选项可恢复丢失或损坏的数据库历史主题(database.history.kafka.topic)。