Kafka与Debezium连接

Debezium 是一个开源的分布式平台，用于捕获数据库的变更数据（CDC，Change Data Capture），并将这些变更数据流式传输到 Kafka 等消息系统中。Debezium 支持多种数据库，包括 MySQL、PostgreSQL、MongoDB 等。

以下是如何将 Kafka 与 Debezium 连接起来的步骤：

1. 安装 Kafka

首先，你需要安装并启动 Kafka。如果你还没有安装 Kafka，可以参考以下步骤：

1. 下载 Kafka： wget https://downloads.apache.org/kafka/3.0.0/kafka_2.13-3.0.0.tgz tar -xzf kafka_2.13-3.0.0.tgz cd kafka_2.13-3.0.0
2. 启动 ZooKeeper： bin/zookeeper-server-start.sh config/zookeeper.properties
3. 启动 Kafka： bin/kafka-server-start.sh config/server.properties

2. 安装 Debezium Connector

Debezium 连接器可以作为 Kafka Connect 的插件运行。你需要下载并安装适用于你数据库的 Debezium 连接器。

1. 下载 Debezium 连接器： wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/1.7.0.Final/debezium-connector-mysql-1.7.0.Final-plugin.tar.gz tar -xzf debezium-connector-mysql-1.7.0.Final-plugin.tar.gz
2. 将连接器放到 Kafka Connect 的插件目录： mkdir -p kafka_2.13-3.0.0/plugins/debezium-connector-mysql mv debezium-connector-mysql/* kafka_2.13-3.0.0/plugins/debezium-connector-mysql/

3. 配置 Kafka Connect

Kafka Connect 是一个用于数据流的框架，支持将数据源连接到 Kafka。你需要配置 Kafka Connect 来使用 Debezium 连接器。

1. 创建 Kafka Connect 配置文件 connect-distributed.properties： bootstrap.servers=localhost:9092 group.id=connect-cluster 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 config.storage.topic=connect-configs offset.storage.topic=connect-offsets status.storage.topic=connect-status plugin.path=plugins
2. 启动 Kafka Connect： bin/connect-distributed.sh config/connect-distributed.properties

4. 配置 Debezium 连接器

你需要创建一个 JSON 文件来配置 Debezium 连接器。以下是一个 MySQL 连接器的示例配置：

1. 创建 register-mysql.json 文件： { "name": "mysql-connector", "config": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "tasks.max": "1", "database.hostname": "localhost", "database.port": "3306", "database.user": "debezium", "database.password": "dbz", "database.server.id": "184054", "database.server.name": "dbserver1", "database.include.list": "inventory", "database.history.kafka.bootstrap.servers": "localhost:9092", "database.history.kafka.topic": "schema-changes.inventory" } }
2. 注册连接器： curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d @register-mysql.json

5. 验证连接器

你可以通过 Kafka 消费者来验证连接器是否正常工作。

1. 启动 Kafka 消费者： bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic dbserver1.inventory.customers --from-beginning

你应该能够看到来自 MySQL 数据库的变更数据流式传输到 Kafka 主题中。