Debezium

[mysqld]
bind-address = 0.0.0.0
binlog_format = ROW
server_id = 1
log_bin = /var/log/mysql/mysql-bin.log

    GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, LOCK TABLES, REPLICATION CLIENT ON *.* TO 'user' IDENTIFIED BY 'password';

key.converter.schemas.enable=false
value.converter.schemas.enable=false

bootstrap.servers=localhost:9092
group.id=connect-cluster
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
#下面这两个只有在json格式的时候才需要配置为false, 其他格式在解析时是需要schema信息的
key.converter.schemas.enable=false  
value.converter.schemas.enable=false
offset.storage.topic=connect-offsets
offset.storage.replication.factor=1
config.storage.topic=connect-configs
config.storage.replication.factor=1
status.storage.topic=connect-status
status.storage.replication.factor=1
offset.flush.interval.ms=10000
listeners=HTTP://:8083
plugin.path=/xxx/kafka/connects

docker run  -d \
    --restart=always \
    --name connect \
    -p 8084:8083 \
    -e GROUP_ID=1 \
    -e CONFIG_STORAGE_TOPIC=docker_connect_configs \
    -e OFFSET_STORAGE_TOPIC=docker_connect_offsets \
    -e STATUS_STORAGE_TOPIC=docker_connect_statuses \
    -e CONNECT_STATUS_STORAGE_REPLICATION_FACTOR=3 \
    -e CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR=3 \
    -e CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR=3 \
    -e CONNECT_OFFSET_FLUSH_INTERVAL_MS=10000 \
    -e OFFSET_FLUSH_INTERVAL_MS=10000 \
    -e BOOTSTRAP_SERVERS=xxxx:9092,xxx:9092 \
    -e CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE=false \
    -e CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE=false \
    -e KEY_CONVERTER=org.apache.kafka.connect.json.JsonConverter \
    -e VALUE_CONVERTER=org.apache.kafka.connect.json.JsonConverter \
    quay.io/debezium/connect:2.4

#具体配置项还可以参考容器内docker_entrypoint.sh

{
    //连接器名称
    "name": "connector-test",
    "config": {
      //连接器类, mysql connector,需根据版本按文档填写
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      //数据库信息
      "database.hostname": "192.168.20.156",
      "database.port": "3306",
      "database.user": "flink",
      "database.password": "flink2023",
      //数据库服务器ID,不能重复
      "database.server.id": "123123",

      //要监控的数据库和表
      "database.include.list": "web",
      "table.include.list": "web.A_ExpertInfo_1,web.W_ListenRec_(.*)",

      //主题前缀,起命名空间作用
      "topic.prefix": "mx1",
      //快照模式,只获取schema,监听在这之后的数据变化
      "snapshot.mode": "schema_only",

      //配置schema历史相关信息,内部使用，不需消费
      "schema.history.internal.kafka.bootstrap.servers": "localhost:9092",
      "schema.history.internal.kafka.topic": "mx.schemahistory.customer",
      //是否包含schema变更事件,在"topic.prefix" 主题里可以看到变更事件
      "include.schema.changes": "true",

      //处理不一致的schema的模式, warn打印warn日志而不是停止任务
      "inconsistent.schema.handling.mode": "warn",
      //不发送墓碑消息
      "tombstones.on.delete": "false",
      
      //定义转换器
      "transforms": "ListenRec,ListenRecPartition",

      //这里配置的是主题路由
      "transforms.ListenRec.type": "io.debezium.transforms.ByLogicalTableRouter",
      "transforms.ListenRec.topic.regex": "(.*)easyweb_new_trans\\.W_ListenRec_(.*)",
      "transforms.ListenRec.topic.replacement": "$1ListenRec",

      //这里配置的是分区路由
      "transforms.ListenRecPartition.type": "io.debezium.transforms.partitions.PartitionRouting",
      "transforms.ListenRecPartition.partition.payload.fields":"change.userid", //这里可以指定before.userid或者after.userid change.userid是自动判断使用before或者after,比如删除后after为空,新建后before为空
      "transforms.ListenRecPartition.partition.topic.num":"12",
      "transforms.ListenRecPartition.predicate":"listen",

      //配置断言,用于指定上面的分区路由,指定如何根据条件将转换应用
      "predicates":"listen",
      "predicates.listen.type":"org.apache.kafka.connect.transforms.predicates.TopicNameMatches",
      "predicates.listen.pattern":"(.*)ListenRec",

      //配置自动创建topic, 分区数和副本数, 注意较低的kafka版本不支持,需要手动创建
      //设置为-1则使用kafka 配置的默认值
      "topic.creation.default.partitions": "1",
      "topic.creation.default.replication.factor": "3",

      //配置自动创建topic的条件,类似transforms
      //还可以配置一些其他topic属性,比如cleanup.policy,retention.ms等
      "topic.creation.listen.partitions":"12",
      "topic.creation.listen.replication.factor":"1",
      "topic.creation.listen.include":"listenrec\\.Listen", //还可以配置exclude
      "topic.creation.groups": "listen",
    }
}

curl -X POST -H "Content-Type: application/json" --data @mysql.json http://localhost:8083/connectors