实时数仓的第一步便是变更数据捕获（CDC）,Debezium就是一款功能非常强大的CDC工具。Debezium是构建于Kafka之上的，将捕获的数据实时的采集到Kafka上
Debezium监控MySQL
监控MySQL的前提是MySQL需要开启binlog日志哦
MySQL开启binlog
MySQL查看是否开启Binlog
代码语言：javascript复制show global variables like 'binlog_format';
show global variables like 'log_bin';
下载MySQL connector
代码语言：javascript复制wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/1.7.1.Final/debezium-connector-mysql-1.7.1.Final-plugin.tar.gzmkdir /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配置可能有所不同 配置文件内容如下
代码语言：javascript复制# 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目录下
代码语言：javascript复制/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请求将配置信息放到请求体就可以了。 配置信息如下：
代码语言：javascript复制{
"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)。 |
