在 RHEL 上安装 Debezium

流程

1. 从 软件下载 站点下载您要使用的 Debezium 连接器或连接器。例如，要将 Debezium 与 MySQL 数据库搭配使用，请下载 Debezium 3.0.8 MySQL Connector。

2. 在您为 Apache Kafka 部署 Streams 的 Red Hat Enterprise Linux 主机上，打开终端窗口并在 /opt/kafka 中创建 connector-plugins 目录（如果它尚不存在）：

   $ sudo mkdir /opt/kafka/connector-plugins

3. 输入以下命令提取您下载到 /opt/kafka/connector-plugins 目录中的 Debezium 连接器存档的内容。

   $ sudo unzip debezium-connector-mysql-3.0.8.Final.zip -d /opt/kafka/connector-plugins

4. 对您要安装的每个连接器重复步骤 1 -3。

5. 在一个终端窗口中，以 kafka 用户身份进行登录：

   $ su - kafka
   $ Password:

6. 如果 Kafka Connect 进程正在运行，请停止。

   1. 输入以下命令检查 Kafka Connect 是否在分布式模式下运行：

      $ jcmd | grep ConnectDistributed

      如果进程正在运行，命令会返回进程 ID，例如：

      18514 org.apache.kafka.connect.cli.ConnectDistributed /opt/kafka/config/connect-distributed.properties

   2. 使用进程 ID 输入 kill 命令来停止进程，例如：

      $ kill 18514

7. 编辑 /opt/kafka/config/ 中的 connect-distributed.properties 文件，并将 plugin.path 的值设置为 Debezium 连接器插件的父目录的位置：

   plugin.path=/opt/kafka/connector-plugins

8. 在分布式模式下启动 Kafka 连接。

   $ /opt/kafka/bin/connect-distributed.sh /opt/kafka/config/connect-distributed.properties

9. 在 Kafka Connect 运行后，使用 Kafka Connect API 注册连接器。
   输入 curl 命令提交 POST 请求，将您在 第 2.2 节 “规划 Debezium 连接器配置” 中指定的连接器配置 JSON 发送到 localhost:8083/connectors 的 Kafka Connect REST API 端点。
   例如：

   curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d '{"name": "inventory-connector", "config":
   {"connector.class": "io.debezium.connector.mysql.MySqlConnector",
   "tasks.max": "1",
   "database.hostname": "mysql",
   "database.port": "3306",
   "database.user": "debezium",
   "database.password": "dbz",
   "database.server.id": "184054",
   "topic.prefix": "dbserver1",
   "table.include.list": "public.inventory",
   "schema.history.internal.kafka.bootstrap.servers": "kafka:9092", "schema.history.internal.kafka.topic": "dbhistory.inventory" }
   }'

   要注册多个连接器，请为每个连接器提交单独的请求。

10. 重启 Kafka Connect 来实现您的更改。

    当 Kafka Connect 启动时，它会从 connector-plugins 目录中加载配置的 Debezium 连接器。

    完成配置后，部署的连接器连接到源数据库，并为每个插入、更新或删除行或文档生成事件。

11. 为每个 Kafka Connect worker 节点重复第 5-10 步。

流程

1. 在主机上的终端窗口中，输入以下命令从 Kafka Connect API 请求连接器列表：

   $ curl -H "Accept:application/json" localhost:8083/connectors/

   查询返回部署的连接器的名称，例如：

   ["inventory-connector"]

2. 在主机上的终端窗口中输入以下命令来查看连接器运行的任务：

   $ curl -i -X GET -H "Accept:application/json" localhost:8083/connectors/inventory-connector

   该命令返回类似以下示例的输出：

   HTTP/1.1 200 OK
   Date: Thu, 06 Feb 2020 22:12:03 GMT
   Content-Type: application/json
   Content-Length: 531
   Server: Jetty(9.4.20.v20190813)
   
   {
     "name": "inventory-connector",
     ...
     "tasks": [
       {
         "connector": "inventory-connector",
         "task": 0
       }
     ]
   }

3. 显示 Kafka 集群中的主题列表。
   在终端窗口中进入 /opt/kafka/bin/ 并运行以下 shell 脚本：

   ./kafka-topics.sh --bootstrap-server=localhost:9092 --list

   Kafka 代理返回连接器创建的主题列表。可用的主题取决于连接器的 snapshot.mode、snapshot.include.collection.list 和 table.include.list 配置属性的设置。默认情况下，连接器会为数据库中的每个非系统表创建一个主题。

4. 查看主题的内容。
   在终端窗口中，导航到 /opt/kafka/bin/，然后运行 kafka-console-consumer.sh shell 脚本来显示上一命令返回的一个主题的内容：
   

   例如：

   ./kafka-console-consumer.sh \
   >     --bootstrap-server localhost:9092 \
   >     --from-beginning \
   >     --property print.key=true \
   >     --topic=dbserver1.inventory.products_on_hand

   对于主题中的每个事件，命令会返回类似以下输出的信息：

   例 2.1. Debezium 更改事件的内容

   {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"}],"optional":false,"name":"dbserver1.inventory.products_on_hand.Key"},"payload":{"product_id":101}}	{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"},{"type":"int32","optional":false,"field":"quantity"}],"optional":true,"name":"dbserver1.inventory.products_on_hand.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"},{"type":"int32","optional":false,"field":"quantity"}],"optional":true,"name":"dbserver1.inventory.products_on_hand.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"string","optional":true,"field":"table"},{"type":"int64","optional":false,"field":"server_id"},{"type":"string","optional":true,"field":"gtid"},{"type":"string","optional":false,"field":"file"},{"type":"int64","optional":false,"field":"pos"},{"type":"int32","optional":false,"field":"row"},{"type":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"field":"transaction"}],"optional":false,"name":"dbserver1.inventory.products_on_hand.Envelope"},"payload":{"before":null,"after":{"product_id":101,"quantity":3},"source":{"version":"3.0.8.Final-redhat-00001","connector":"mysql","name":"inventory_connector_mysql","ts_ms":1638985247805,"snapshot":"true","db":"inventory","sequence":null,"table":"products_on_hand","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":156,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1638985247805,"transaction":null}}

   在前面的示例中，有效负载 值显示连接器快照从表 inventory.products_on_hand 生成一个读取("op" ="r")事件。product_id 记录的 "before" 状态为 null，表示记录没有之前的值。"after" 状态对于 product_id 为 101 的项目的 quantity 显示为 3。

流程

1. 从 Software Downloads 下载您要替换的 Debezium 连接器插件的副本。

2. 将 Debezium 连接器存档的内容提取到 /opt/kafka/connector-plugins 目录中。

   $ sudo unzip debezium-connector-mysql-3.0.8.Final.zip -d /opt/kafka/connector-plugins

3. 重启 Kafka Connect。