在 RHEL 上安装 Debezium

前提条件

• 部署源数据库，Debezium 连接器可以访问数据库。

• 您知道以下信息，连接器需要访问源数据库：

  • 数据库主机的名称或 IP 地址。
  • 用于连接数据库的端口号。
  • 连接器可用于登录数据库的帐户的名称。
  • 数据库用户帐户的密码。
  • 数据库的名称。
  • 您希望连接器的表的名称捕获信息。
  • 希望连接器要发出事件的 Kafka 代理名称。
  • 您希望连接器发送到的 Kafka 主题的名称，以发送数据库历史记录信息。

流程

• 以 JSON 格式指定您要应用到 Debezium 连接器的配置。
  

  以下示例显示了 Debezium MySQL 连接器的简单配置：

  {
    "name": "inventory-connector",  

  1

  
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector", 

  2

  
      "tasks.max": "1",  

  3

  
      "database.hostname": "mysql",  

  4

  
      "database.port": "3306", 

  5

  
      "database.user": "debezium", 

  6

  
      "database.password": "dbz", 

  7

  
      "database.server.id": "184054",  

  8

  
      "database.server.name": "dbserver1",
      "database.include.list": "public.inventory",  

  9

  
      "database.history.kafka.bootstrap.servers": "kafka:9092",  

  10

  
      "database.history.kafka.topic": "dbhistory.inventory"  

  11

  
    }
  }

  Copy to Clipboard Toggle word wrap

  1

  使用 Kafka Connect 集群注册的连接器名称。

  2

  连接器类的名称。

  3

  可同步运行的任务数量。次仅应操作一个任务。

  4

  主机数据库实例的主机名或 IP 地址。

  5

  数据库实例的端口号。

  6

  Debezium 连接到数据库的用户帐户的名称。

  7

  数据库用户帐户的密码。

  8

  唯一的服务器 ID 和名称。服务器名称是可以访问 Debezium 的数据库服务器或服务器集群的逻辑标识符。此名称的 Debezium 将这个名称前缀到它创建的每个 Kafka 主题。

  9

  连接器从中捕获更改事件的表列表。

  10

  连接器发送数据库模式历史记录的 Kafka 代理名称。指定代理还会接收连接器发布的更改事件。

  11

  存储 schema 历史记录的 Kafka 主题名称。
  在连接器重启后，连接器会从它停止的点中恢复数据库日志，为在离线时发生的任何事务发送事件。在连接器写入对 Kafka 的未读取事务更改事件之前，它会检查 schema 历史记录，然后在原始事务发生时应用效果的 schema。

附加信息

• 有关您可以为各种连接器设置的配置属性的信息，请参阅 Debezium 用户指南中的 连接器的部署文档。

前提条件

• 要部署 Debezium 的主机环境在 受支持的配置 中运行 Red Hat Enterprise Linux、AMQ Streams 和 Java。
  

  • 有关如何安装 AMQ Streams 的详情，请参考 安装 AMQ Streams。
  • 有关如何安装包含单一 ZooKeeper 节点的基本、非生产 AMQ Streams 集群以及单个 Kafka 节点的详情，请参考 运行单一节点 AMQ Streams 集群。

• 在主机上具有管理特权(sudo 访问权限)。
• Apache ZooKeeper 和 Apache Kafka 代理正在运行。
• Kafka Connect 以 分布式模式运行，而不是在独立模式下运行。
• 您知道安装了 AMQ Streams 时创建的 kafka 用户的凭证。
• 部署源数据库，您部署 Debezium 可以访问数据库的主机。
• 您知道 如何配置连接器。

其他资源

• 添加连接器插件

前提条件

• 根据 第 2.3 节 “在 Red Hat Enterprise Linux 中使用 AMQ Streams 部署 Debezium” 中的说明，您在 Red Hat Enterprise Linux 上部署了连接器。

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

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

     Copy to Clipboard Toggle word wrap

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

     ["inventory-connector"]

     Copy to Clipboard Toggle word wrap

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

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

     Copy to Clipboard Toggle word wrap

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

     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
         }
       ]
     }

     Copy to Clipboard Toggle word wrap

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

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

     Copy to Clipboard Toggle word wrap

     Kafka 代理返回连接器创建的主题列表。可用主题取决于连接器的快照. mode、snapshot. include.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=inventory_connector_mysql.inventory.products_on_hand

     Copy to Clipboard Toggle word wrap

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

     例 2.1. Debezium 更改事件的内容

     {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"}],"optional":false,"name":"inventory_connector_mysql.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":"inventory_connector_mysql.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":"inventory_connector_mysql.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":"inventory_connector_mysql.inventory.products_on_hand.Envelope"},"payload":{"before":null,"after":{"product_id":101,"quantity":3},"source":{"version":"1.9.7.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}}

     Copy to Clipboard Toggle word wrap

     在前面的示例中，有效负载 值显示连接器快照从表 清单中生成读取(op" ="r")事件。products_on_hand.product_id 记录的 "前" 状态 为空，表示记录不存在之前的值。"after" 状态显示带有 product_id 101 的项目的 3 数量。