第 3 章 sink 连接器

流程

1. 登录 OpenShift 集群。

2. 为连接器创建 Debezium KafkaConnect 自定义资源(CR)，或修改现有的资源。
   例如，使用名称 dbz-jdbc-connect.yaml 创建一个 KafkaConnect CR，用于指定 注解和 镜像 属性，如以下摘录所示：在以下示例中，自定义资源被配置为下载以下工件：

   • Debezium JDBC 连接器存档。

   • 连接到 Oracle 或 Db2 sink 数据库所需的 JDBC 驱动程序。您可以为其他 sink 目的地省略此条目。

     定义包含 Debezium 连接器的 KafkaConnect 自定义资源的 dbz-jdbc-connect.yaml 文件

     apiVersion: kafka.strimzi.io/v1beta2
     kind: KafkaConnect
     metadata:
       name: debezium-kafka-connect-cluster
       annotations:
         strimzi.io/use-connector-resources: "true"  //  

     1

     
     spec:
       version: 3.9.0
       replicas: 1
       bootstrapServers: debezium-kafka-cluster-kafka-bootstrap:9093
       build:  

     2

     
         output:  

     3

     
           type: imagestream  

     4

     
           image: debezium-streams-connect:latest
         plugins:   

     5

     
           - name: debezium-jdbc-connector
             artifacts:
               - type: zip  

     6

     
                 url: https://maven.repository.redhat.com/ga/io/debezium/debezium-connector-jdbc/3.0.8.Final-redhat-00004/debezium-connector-jdbc-3.0.8.Final-redhat-00004-plugin.zip  

     7

     
               - type: jar
                 url: https://repo1.maven.org/maven2/com/oracle/ojdbc/ojdbc11/21.15.0.0/ojdbc11-21.15.0.0.jar  

     8

     
       ...

     Copy to Clipboard Toggle word wrap

     Expand

     表 3.5. Kafka Connect 配置设置的描述

     项	描述
     1	将 strimzi.io/use-connector-resources 注解设置为 "true"，以便 Cluster Operator 使用 KafkaConnector 资源在此 Kafka Connect 集群中配置连接器。
     2	spec.build 配置指定存储构建镜像的位置，并列出镜像中包含的插件，以及插件工件的位置。
     3	build.output 指定存储新构建的镜像的 registry。
     4	指定镜像输出的名称和镜像名称。output.type 的有效值为 docker，可推送到容器 registry （如 Quay）或 镜像流 （用于将镜像推送到内部 OpenShift ImageStream）。要使用 ImageStream，必须将 ImageStream 资源部署到集群中。有关在 KafkaConnect 配置中指定 build.output 的更多信息，请参阅 {Name configuringStreamsOpenShift} 中的 Apache Kafka Build schema 参考。
     5	插件配置 列出了您要包含在 Kafka Connect 镜像中的所有连接器。对于列表中的每个条目，指定一个插件名称，以及有关构建连接器所需的工件的信息。另外，对于每个连接器插件，您可以包括要用于连接器的其他组件。例如，您可以添加 Service Registry 工件或 Debezium 脚本组件。
     6	artifacts.type 的值指定 artifacts.url 中指定的工件的文件类型。有效类型是 zip、tgz、或 jar。Debezium 连接器存档以 .zip 文件格式提供。JDBC 驱动程序文件采用 .jar 格式。type 值必须与 url 字段中引用的文件类型匹配。
     7	artifacts.url 的值指定 HTTP 服务器的地址，如 Maven 存储库，用于存储连接器工件的文件。OpenShift 集群必须有权访问指定的服务器。
     8	（仅用于 Db2 或 Oracle sinks）指定 Maven Central 中的 JDBC JDBC 驱动程序的位置。Debezium 连接到这些数据库所需的驱动程序不包括在 Debezium 连接器存档中。 这个示例为 Oracle Database JDBC 驱动程序提供 Maven URL。Db2 JDBC 驱动程序位于以下 Maven 位置 ：https://repo1.maven.org/maven2/com/ibm/db2/jcc/11.5.9.0/jcc-11.5.9.0.jar

     Show more

3. 输入以下命令将 KafkaConnect 构建规格应用到 OpenShift 集群：

   oc create -f dbz-jdbc-connect.yaml

   Copy to Clipboard Toggle word wrap

   根据自定义资源中指定的配置，Streams Operator 准备要部署的 Kafka Connect 镜像。
   构建完成后，Operator 将镜像推送到指定的 registry 或 ImageStream，并启动 Kafka Connect 集群。您在配置中列出的连接器工件在集群中可用。

4. 创建一个 KafkaConnector 资源来定义您要部署的每个连接器的实例。
   例如，创建以下 KafkaConnector CR，并将它保存为 orders-to-postgresql-jdbc-connector.yaml

   orders-to-postgresql-jdbc-connector.yaml 文件，该文件为 Debezium 连接器定义 KafkaConnector 自定义资源

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: orders-topic-to-postgresql-via-jdbc-sink-connector 

   1

   
     labels: strimzi.io/cluster: debezium-kafka-connect-cluster
   spec:
     class: io.debezium.connector.jdbc.JdbcSinkConnector  

   2

   
     tasksMax: 1  

   3

   
      config:  

   4

   
       connection.url: jdbc://postgresql://<host>:<port>/<database-name> 

   5

   
       connection.username: <database-user>  

   6

   
       connection.password: <database-pwd>  

   7

   
       insert.mode: upsert
       delete.enabled: true
       topics: orders 

   8

   
       primary.key.mode: record_key
       schema.evolution: basic
       use.time.zone: UTC

   Copy to Clipboard Toggle word wrap

   Expand

   表 3.6. 连接器配置设置的描述

   项	描述
   1	要注册到 Kafka Connect 集群的连接器名称。
   2	连接器类的名称。
   3	可同时操作的任务数量。
   4	连接器的配置。
   5	sink 数据库的 JDBC 连接 URL。URL 指定端口号以及连接到数据库所需的任何身份验证属性。例如，jdbc:oracle:thin:@myhost.example.com:1521/myservice
   6	Debezium 用来连接到数据库的帐户名称。
   7	Debezium 用来连接到数据库用户帐户的密码。
   8	指定连接器读取的 Kafka 主题的逗号分隔列表。每个主题中的事件被流传输到 sink 数据库中名称相同的表。

   Show more

5. 运行以下命令来创建连接器资源：

   oc create -n <namespace> -f <kafkaConnector>.yaml

   Copy to Clipboard Toggle word wrap

   例如，

   oc create -n debezium -f jdbc-inventory-connector.yaml

   Copy to Clipboard Toggle word wrap

   连接器注册到 Kafka Connect 集群，并开始针对 KafkaConnector CR 中的 spec.config.database.dbname 指定的数据库运行。连接器 pod 就绪后，Debezium 正在运行。

流程

1. 在 Kafka Connect 上创建 Debezium JDBC 连接器容器：

   1. 创建一个 Containerfile，它使用 registry.redhat.io/amq-streams/kafka-39-rhel9:2.9.0 作为基础镜像。例如，在终端窗口中输入以下命令：
      

1. 从您在上一步中创建的 debezium-jdbc-connector-container.yaml Containerfile 构建容器镜像。在包含该文件的目录中，打开终端窗口并输入以下命令之一：

   podman build -t debezium-jdbc-connector-container:latest .

   Copy to Clipboard Toggle word wrap

   docker build -t debezium-jdbc-connector-container:latest .

   Copy to Clipboard Toggle word wrap

   前面的命令使用名称 debezium-jdbc-connector-container 构建容器镜像。

2. 将自定义镜像推送到容器 registry，如 quay.io 或内部容器 registry。容器镜像仓库必须可供您要部署镜像的 OpenShift 实例使用。输入以下命令之一：

   podman push <myregistry.io>/debezium-jdbc-connector-container:latest

   Copy to Clipboard Toggle word wrap

   docker push <myregistry.io>/debezium-jdbc-connector-container:latest

   Copy to Clipboard Toggle word wrap

3. 创建新的 Debezium Oracle KafkaConnect 自定义资源(CR)。例如，使用名称 dbz-connect.yaml 创建 KafkaConnect CR，用于指定 注解和 镜像 属性。以下示例显示了描述 KafkaConnect 自定义资源的 dbz-connect.yaml 文件摘录。
   

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: debezium-kafka-connect-cluster
     annotations:
       strimzi.io/use-connector-resources: "true" 

   1

   
   spec:
     image: debezium-jdbc-connector-container 

   2

   
   
     ...

   Copy to Clipboard Toggle word wrap

1. 输入以下命令将 KafkaConnect CR 应用到 OpenShift Kafka Connect 环境：

   oc create -f dbz-connect.yaml

   Copy to Clipboard Toggle word wrap

   该命令添加一个 Kafka Connect 实例，用于指定为运行 Debezium 连接器而创建的镜像的名称。

   1. 创建一个 KafkaConnector 自定义资源，用于配置 Debezium JDBC 连接器实例。

      您可以在 .yaml 文件中配置 Debezium JDBC 连接器，该文件指定连接器的配置属性。

      以下示例显示了 dbz-connect.yaml 文件的摘录，它为 KafkaConnect 自定义资源设置几个关键属性。
      连接器在端口 5432 上建立到 PostgreSQL 服务器 sink 的 JDBC 连接。
      
      有关可用连接器属性的完整范围的详情，请参阅 Debezium JDBC 连接器配置属性的描述。

      例 3.1. jdbc-connector.yaml

      apiVersion: kafka.strimzi.io/v1beta2
      kind: KafkaConnector
      metadata:
        labels:
        strimzi.io/cluster: my-connect-cluster 

      1

      
        name: orders-topic-to-postgresql-via-jdbc-sink-connector 

      2

      
      ...
      spec:
        class: io.debezium.connector.jdbc.JdbcSinkConnector 

      3

      
      ...
        config:
          connection.url: jdbc:<_jdbc_>://<_database-host_>:5432/<_database-name_>  

      4

      
          connection.username: "<_database-user_>" 

      5

      
          connection.password: "<_database-password_>" 

      6

      
          topics: orders 

      7

      Copy to Clipboard Toggle word wrap

1. 使用 Kafka Connect 创建连接器实例。例如，如果您在 jdbc-connector.yaml 文件中保存 KafkaConnector 资源，您将运行以下命令：

   oc apply -f jdbc-connector.yaml

   Copy to Clipboard Toggle word wrap

   前面的命令注册 orders-topic-to-postgresql-via-jdbc-sink-connector。根据 KafkaConnector CR 中指定的，连接器启动并开始从 订购 主题中读取。