Debizum Kafka Connect实现Mysql CDC功能

罪恶克星  金牌会员 | 2025-1-6 10:43:53 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 867|帖子 867|积分 2601

Debizum Kafka Connect架构

最常见的是通过Apache Kafka Connect部署Debezium。Kafka Connect是一个框架和运行时，用于实现和操作:


  • 源毗连器 （source connect）
  • 吸收毗连器（sink connect），将记录从Kafka主题流传到其他系统
    下图体现了基于Debezium的变动数据捕获管道的架构:

1.Mysql数据库预备


  • 配置mysql的binlog日志必要开启！！！！
  1. show variables like 'log_%';
复制代码
假如 log_bin 字段是 OFF,可通过下列配置文件my.cnf开启.
  1.    # [必须配置]服务ID
  2.    server-id=1
  3.    # 建议配置在[datadir]目录下, binlog 配置 只要配置了log_bin地址 就会开启
  4.    log_bin = /data/mysql/mysql_bin
  5.    # 日志存储天数 默认0 永久保存
  6.    # 如果数据库会定期归档，建议设置一个存储时间不需要一直存储binlog日志，理论上只需要存储归档之后的日志
  7.    expire_logs_days = 30
  8.    # [mysql8.0被废弃]binlog最大值
  9.    max_binlog_size = 1024M
  10.    # 规定binlog的格式，binlog有三种格式statement、row、mixad，默认使用statement，
  11.    # 建议使用row格式
  12.    binlog_format = ROW
  13.    # [重点]在提交n次事务后，进行binlog的落盘，0为不进行强行的刷新操作，而是由文件系统控制刷新日志文件，如果是在线交易和账有关的数据建议设置成1，如果是其他数据可以保持为0即可
  14.    sync_binlog = 1
复制代码
然后再重启,systemctl restart mysqld或systemctl restart mysql.service
2.Debezium环境预备

1）zookeeper+kafka+kafka ui
  1. version: "2"
  3. services:
  4.     zookeeper:
  5.         image: confluentinc/cp-zookeeper:latest
  6.         container_name: debezium-zookeeper
  7.         environment:
  8.             ZOOKEEPER_CLIENT_PORT: 2181
  9.             ZOOKEEPER_TICK_TIME: 2000
  10.             ZK_SERVER_HEAP: "-Xmx256M -Xms256M"
  11.         ports:
  12.             - 22181:2181
  13.     kafka:
  14.         image: confluentinc/cp-kafka:latest
  15.         container_name: debezium-kafka-broker
  16.         depends_on:
  17.             - zookeeper
  18.         ports:
  19.             - 29092:29092
  20.         environment:
  21.             KAFKA_BROKER_ID: 1
  22.             KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
  23.             KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://192.168.14.107:29092
  24.             KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
  25.             KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
  26.             KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
  27.             KAFKA_HEAP_OPTS: "-Xmx256M -Xms256M"
  28.     init-kafka:
  29.         image: confluentinc/cp-kafka:latest
  30.         depends_on:
  31.           - kafka
  32.         entrypoint: [ '/bin/sh', '-c' ]
  33.         command: |
  34.           "
  35.           # blocks until kafka is reachable
  36.           kafka-topics --bootstrap-server kafka:9092 --list
  37.    
  38.           echo -e 'Creating kafka topics'
  39.           kafka-topics --bootstrap-server kafka:9092 --create --if-not-exists --topic product-update-fail --replication-factor 1 --partitions 1
  40.           kafka-topics --bootstrap-server kafka:9092 --create --if-not-exists --topic order-created --replication-factor 1 --partitions 1
  41.           kafka-topics --bootstrap-server kafka:9092 --create --if-not-exists --topic product-update-successfully --replication-factor 1 --partitions 1
  42.    
  43.           echo -e 'Successfully created the following topics:'
  44.           kafka-topics --bootstrap-server kafka:9092 --list
  45.           "
  46.     kafka-ui:
  47.         container_name: debezium-kafka-ui
  48.         image: provectuslabs/kafka-ui:latest
  49.         ports:
  50.           - 9000:8080
  51.         depends_on:
  52.           - zookeeper
  53.           - kafka
  54.         environment:
  55.           KAFKA_CLUSTERS_0_NAME: loan-kafka-ui
  56.           KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
  57.           KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181
  59.      
  60. networks:
  61.   debezium_default:
  62.     driver: bridge
复制代码
2）debezium
偶然间可以搞一下 [Debezium UI].(https://debezium.io/documentation/reference/2.7/operations/debezium-ui.html)
  1. version: "2"
  3. services:
  4.   connect:
  5.     image: quay.io/debezium/connect:2.7.3.Final
  6.     networks:
  7.       - debezium_default
  8.     ports:
  9.      - 7750:8083
  10.     environment:
  11.     ## kafka链接地址
  12.      - BOOTSTRAP_SERVERS=kafka:9092
  13.      - GROUP_ID=1
  14.      - CONFIG_STORAGE_TOPIC=mx.connect_configs
  15.      - OFFSET_STORAGE_TOPIC=mx.connect_offsets
  16.      - STATUS_STORAGE_TOPIC=mx.connect_statuses
  17.      - HEAP_OPTS=-Xms256M -Xmx512M
  18. #  debezium-ui:
  19. #    image: quay.io/debezium/debezium-ui:2.5
  20. #    networks:
  21. #      - debezium_default
  22. #    ports:
  23. #      - 7780:8080
  24. #    environment:
  25. #      - KAFKA_CONNECT_URIS=http://connect:8083
  27. networks:
  28.   debezium_default:
  29.      #添加到上面已有的网络中！
  30.      externl: true
复制代码
3.创建Debuzium毗连

debuzium 安装乐成后，想要捕捉mysql数据库变动，必要哀求debuzium 对外暴露的接口创建毗连器，执行下面创建一个source的connector的哀求，执行乐成，则会正确返回，再执行获取指定connector状态的哀求，如果得到一下结果，则阐明毗连器创建乐成。
3.1.connect暴露底子API接口

  1. #检测活跃的connect
  2. GET  http://192.168.204.134:7750/connectors/
复制代码
  1. #创建新的connect
  2. POST  http://192.168.204.134:7750/connectors
  3. Content-Type: application/json
  4. body:{略...}
复制代码
  1. #删除connect
  2. DELETE http://192.168.204.134:7750/connectors/${connect-name}
复制代码
  1. #获取详情
  2. http://192.168.204.134:7750/connectors/${connect-name}/config
复制代码
3.2.创建source connect

  1. POST  http://192.168.204.134:7750/connectors
  2. Content-Type: application/json
  4. {
  5.   "name": "connector-test-db",
  6.   "config": {
  7.     "connector.class": "io.debezium.connector.mysql.MySqlConnector",
  8.     "database.hostname": "192.168.204.134",
  9.     "database.port": "3306",
  10.     "database.user": "root",
  11.     "database.password": "123456",
  12.     "database.server.id": "10001",
  13.     "database.server.name": "debezium-server-test",
  14.     "database.serverTimezone":"UTC",
  15.     "topic.prefix": "debezium-server-test",
  16.     "time.precision.mode": "connect"
  17. ,
  18.     "database.include.list": "spring-debezium-demo",
  19.     "table.include.list": "spring-debezium-demo.sys_user",
  20.     "snapshot.mode": "schema_only",
  21.     "database.history.kafka.bootstrap.servers": "kafka:9092",
  22.     "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
  23.     "schema.history.internal.kafka.topic": "debezium-server-test.schemahistory.spring-debezium-demo",
  24.     "database.history.kafka.topic": "debezium-server-test.debezium-server_history_schema",
  25.     "include.schema.changes": "true",
  26.     "transforms":"unwrap",
  27.     "transforms.unwrap.type":"io.debezium.transforms.ExtractNewRecordState",
  28.     "transforms.unwrap.drop.tombstones":"true",
  29.     "transforms.unwrap.delete.handling.mode":"none",
  30.     "transforms.unwrap.operation.header":"false"
  31.   }
  32. }
复制代码
参数分析:


  • name：毗连器的唯一名称，在 Kafka Connect 集群中应保持唯一性。
  • config：包含毗连器的具体配置项。

    • connector.class：指定使用 Debezium 的 MySQL 毗连器类。
    • database.hostname： 目的 MySQL 数据库服务器的主机地址。
    • database.port：目的 MySQL 数据库服务器的端标语，默认 MySQL 端口为 3306。
    • database.user： 用于毗连 MySQL 数据库的用户名。
    • database.password：对应用户名的密码。
    • database.server.id：毗连器在 MySQL 集群中的唯一服务器 ID，克制与其他 MySQL 服务器或毗连器冲突。
    • database.server.name： 逻辑服务器名称，用作天生的 Kafka 主题的前缀，标识捕获数据的来源。
    • topic.prefix：指定 Kafka 主题的前缀，实际的主题名称通常为 {topic.prefix}.{database}.{table}。应该和上面的${database.server.name}划一。
    • database.include.list：指定要捕获的数据库名称，毗连器仅捕获该数据库中的更改。
    • table.include.list：指定要捕获的表，格式为 {database}.{table}，仅捕获该表的更改。
    • snapshot.mode： 界说初始快照模式：
    • schema_only：仅捕获数据库模式（布局），不包含数据。其他选项包括 initial（捕获模式和数据）等。
    • database.history.kafka.bootstrap.servers：指定用于存储数据库历史记录的 Kafka broker 地址。
    • schema.history.internal.kafka.bootstrap.servers： 指定用于存储内部模式历史的 Kafka broker 地址。
    • schema.history.internal.kafka.topic：存储内部模式历史的 Kafka 主题名称。
    • database.history.kafka.topic： 存储数据库历史信息的 Kafka 主题名称。
    • include.schema.changes：指定是否将数据库模式更改（DDL 语句）作为事故写入 Kafka 主题。true 表现写入。

3.2.创建sink connect

Connect容器自带有Debezium的官方Source Connector：


  • debezium-connector-db2
  • debezium-connector-mysql
  • debezium-connector-postgres
  • debezium-connector-vitess
  • debezium-connector-mongodb
  • debezium-connector-oracle
  • debezium-connector-sqlserver
    sink connect必要从confluent 官网下载kafka-connect-jdbc包
  1. # docker容器中新建kafka-connect-jdbc目录
  2. docker exec 容器id mkdir /kafka/connect/kafka-connect-jdbc
  3. # 下载jar包到本地
  4. wget https://packages.confluent.io/maven/io/confluent/kafka-connect-jdbc/10.7.4/kafka-connect-jdbc-10.7.4.jar
  5. # 拷贝jar包到docker容器
  6. docker cp kafka-connect-jdbc-10.7.4.jar 容器id:/kafka/connect/kafka-connect-jdbc
  7. # 重启connect容器
  8. docker restart 容器id
  11. #【踩坑】注意这个包是没有驱动，需要把容器内debezium-connector-mysql文件夹mysq-connector复制到kafka-connect-jdbc下面
复制代码
创建一个Sink connect目的是下游Mysql
  1. POST  http://192.168.204.134:7750/connectors
  2. Content-Type: application/json
  3. {
  4.   "name": "sink-sys-user-mysql",
  5.   "config": {
  6.     "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
  7.     "tasks.max": "1",
  8.     "topics": "debezium-server-test.spring-debezium-demo.sys_user",
  9.     "connection.url": "jdbc:mysql://192.168.204.134:3306/debezium-sink-db?useSSL=false&serverTimezone=UTC",
  10.     "connection.user": "root",
  11.     "connection.password": "123456",
  12.     "insert.mode": "upsert",
  13.     "auto.create": "true",
  14.     "auto.evolve": "true",
  15.     "pk.fields": "ID",
  16.     "pk.mode": "record_key",
  17.     "table.name.format": "sys_user",
  18.     "fields.whitelist": "ID,NAME,USERNAME,PASSWORD,GENDER,REMARK,STATUS,CREATE_TIME,CREATE_ID,UPDATE_TIME,UPDATE_ID",
  19.     "delete.enabled": "true",
  20.     "batch.size": "500",
  21.     "transforms": "convertTimestamps",
  22.     "transforms.convertTimestamps.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
  23.     "transforms.convertTimestamps.field": "CREATE_TIME,UPDATE_TIME",
  24.     "transforms.convertTimestamps.target.type": "Timestamp",
  25.     "transforms.convertTimestamps.format": "yyyy-MM-dd HH:mm:ss"
  26.   }
  27. }
  29. #【踩坑】一定要注意数据库，表字段的大小写
复制代码
生产环境不可能一个表一个connect，可以使用topic正则来动态适配，(注意先在source connect配置好table.include.list)
  1. #取消直接命名
  2. #"table.name.format": "sys_user"
  3. #监听的"topics"改为"topics.regex"
  4. #改为下面的
  5. "topics.regex": "debezium-server-test.spring-debezium-demo.*",
  6. "transforms":"routeTopics",
  7. "transforms.routeTopics.type":"org.apache.kafka.connect.transforms.RegexRouter",
  8. "transforms.routeTopics.regex":"^(.*)\\.spring-debezium-demo\\.(.*)$",
  9. "transforms.routeTopics.replacement":"$2",
  10. "table.name.format":"${topic}"
复制代码
【踩坑】：mysql CDC 关于时间 Incorrect datetime value: ‘1733827592000’ for column ‘CREATE_TIME’ at row 1
办理办法：
方法一：（可能无效）
source添加：
  1. "time.precision.mode": "connect"
复制代码
sink添加：
  1. "transforms":"TimestampConverter"
  2. "transforms.TimestampConverter.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value"
  3. "transforms.TimestampConverter.format": "yyyy-MM-dd HH:mm:ss"
  4. "transforms.TimestampConverter.target.type": "Timestamp"
  5. "transforms.TimestampConverter.field": "time"
复制代码
方法二：（有效）
source添加：
  1. "database.serverTimezone":"UTC",
  2. "time.precision.mode":"connect",
  3. "transforms":"unwrap",
  4. "transforms.unwrap.type":"io.debezium.transforms.ExtractNewRecordState",
  5. "transforms.unwrap.drop.tombstones":"true",
  6. "transforms.unwrap.delete.handling.mode":"none",
  7. "transforms.unwrap.operation.header":"false"
复制代码
sink添加：
  1. "transforms.TimestampConverter.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
  2. "transforms.TimestampConverter.target.type": "Timestamp",
  3. "transforms.TimestampConverter.field": "create_time,his_create_time,his_update_time"
复制代码
4.运行结果




5.总结

更多资料查阅官方网站：https://debezium.io/documentation/reference/3.0/connectors/jdbc.html
国内的文档较少，CDC过程中发现，就算配置了auto.create是无效的。并不会在sink connect中创建同步的表。特殊要注意，source端和sink的一些数据类型，比如datetime，bigdecimal，json等对应debezium的数据类型。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

罪恶克星

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

存储 挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表