KnowForge/Debezium 简体中文
运营

在 OpenShift 上运行

qianmoQqianmoQ· 更新于 2026-09-28· 阅读 22 分钟· 0 次阅读
登录后可跨设备保存划线和私人笔记登录

在 OpenShift 上部署 Debezium

本过程用于在 Red Hat 的 OpenShift 容器平台上设置 Debezium 连接器。若要在 OpenShift 上进行开发或测试，可以使用 CodeReady Containers。

想要更快上手，可以试试 Debezium 在线学习场景。它会为你单独启动一个 OpenShift 集群，让你在几分钟内就能在浏览器中开始使用 Debezium。

前提条件

为使容器与集群上的其他工作负载保持隔离，请为 Debezium 创建一个专用项目。在本文档的后续内容中，将使用 debezium-example 命名空间：

$ oc new-project debezium-example

部署 Strimzi Operator

在部署 Debezium 时，我们将使用 Strimzi 项目，它用于在 OpenShift 集群上管理 Kafka 的部署。安装 Strimzi 最简单的方式是从 OperatorHub 安装 Strimzi operator。在 OpenShift 界面中导航到“OperatorHub”标签页，选择“Strimzi”，然后点击“install”按钮。

openshift strimzi operator

如果你更喜欢命令行工具，也可以通过以下方式安装 Strimzi operator：

$ cat << EOF | oc create -f -
apiVersion: operators.coreos.com/v1alpha1
kind: Subscription
metadata:
  name: my-strimzi-kafka-operator
  namespace: openshift-operators
spec:
  channel: stable
  name: strimzi-kafka-operator
  source: operatorhubio-catalog
  sourceNamespace: olm
EOF

为数据库创建 Secret

稍后在部署 Debezium Kafka 连接器时，我们需要为连接器提供用户名和密码，使其能够连接到数据库。出于安全考虑，良好的实践是不直接提供这些凭据，而是将它们保存在单独的安全位置。OpenShift 为此提供了 Secret 对象。除了创建 Secret 对象本身之外，我们还必须创建一个角色和一个角色绑定，以便 Kafka 能够访问这些凭据。

我们先来创建 Secret 对象：

$ cat << EOF | oc create -f -
apiVersion: v1
kind: Secret
metadata:
  name: debezium-secret
  namespace: debezium-example
type: Opaque
data:
  username: ZGViZXppdW0=
  password: ZGJ6
EOF

username 和 password 中包含用于连接 MySQL 数据库的 base64 编码凭据（debezium/dbz），该数据库将在稍后部署。

现在，我们可以创建一个角色，该角色引用上一步中创建的 Secret：

$ cat << EOF | oc create -f -
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: connector-configuration-role
  namespace: debezium-example
rules:
- apiGroups: [""]
  resources: ["secrets"]
  resourceNames: ["debezium-secret"]
  verbs: ["get"]
EOF

我们还需要将该角色绑定到 Kafka Connect 集群服务账号，以便 Kafka Connect 能够访问该密钥：

$ cat << EOF | oc create -f -
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: connector-configuration-role-binding
  namespace: debezium-example
subjects:
- kind: ServiceAccount
  name: debezium-connect-cluster-connect
  namespace: debezium-example
roleRef:
  kind: Role
  name: connector-configuration-role
  apiGroup: rbac.authorization.k8s.io
EOF

服务账户将由 Strimzi 在我们部署 Kafka Connect 时创建。服务账户的名称格式为 $KafkaConnectName-connect。稍后我们将创建名为 debezium-connect-cluster 的 Kafka Connect 集群，因此这里使用 debezium-connect-cluster-connect 作为 subjects.name。

部署 Apache Kafka

接下来，部署一个 Kafka 集群（通过 NodePool 部署 3 个节点）：

$ cat << EOF | oc create -n debezium-example -f -
---
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaNodePool
metadata:
  name: debezium-cluster-node-pool
  labels:
    strimzi.io/cluster: debezium-cluster
spec:
  replicas: 3
  roles:
    - broker
    - controller
  storage:
    type: ephemeral
---
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: debezium-cluster
  annotations:
    strimzi.io/node-pools: enabled
    strimzi.io/kraft: enabled
spec:
  kafka:
    version: "3.9.0"
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
      - name: external
        port: 9094
        type: nodeport
        tls: false
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      default.replication.factor: 3
      min.insync.replicas: 2
  entityOperator:
    topicOperator: {}
    userOperator: {}
---
EOF
  • 等待其就绪：
$ oc wait kafka/debezium-cluster --for=condition=Ready --timeout=300s

部署数据源

下面将以 MySQL 作为数据源。除了运行一个 MySQL 的 Pod 之外，还需要一个指向数据库 Pod 本身的 Service。例如，可以按如下方式创建它：

$ cat << EOF | oc create -f -
apiVersion: v1
kind: Service
metadata:
  name: mysql
spec:
  ports:
  - port: 3306
  selector:
    app: mysql
  clusterIP: None
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mysql
spec:
  selector:
    matchLabels:
      app: mysql
  strategy:
    type: Recreate
  template:
    metadata:
      labels:
        app: mysql
    spec:
      containers:
      - image: quay.io/debezium/example-mysql:3.6
        name: mysql
        env:
        - name: MYSQL_ROOT_PASSWORD
          value: debezium
        - name: MYSQL_USER
          value: mysqluser
        - name: MYSQL_PASSWORD
          value: mysqlpw
        ports:
        - containerPort: 3306
          name: mysql
EOF

部署 Debezium 连接器

要部署 Debezium 连接器，需要先部署一个包含所需连接器插件的 Kafka Connect 集群，然后再实例化连接器本身。第一步，必须创建一个包含插件的 Kafka Connect 容器镜像。如果你已经在镜像仓库中构建并提供了容器镜像，可以跳过这一步。本文将以 MySQL 连接器为例进行说明。

创建 Kafka Connect 集群

我们将再次使用 Strimzi 来创建 Kafka Connect 集群。Strimzi 也可以替我们构建并推送所需的容器镜像。实际上，这两项任务可以合并在一起，构建容器镜像的说明可以直接写在 KafkaConnect 对象的规范中：

$ cat << EOF | oc create -f -
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: debezium-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  version: 3.1.0
  replicas: 1
  bootstrapServers: debezium-cluster-kafka-bootstrap:9092
  config:
    config.providers: secrets
    config.providers.secrets.class: io.strimzi.kafka.KubernetesSecretConfigProvider
    group.id: connect-cluster
    offset.storage.topic: connect-cluster-offsets
    config.storage.topic: connect-cluster-configs
    status.storage.topic: connect-cluster-status
    # -1 means it will use the default replication factor configured in the broker
    config.storage.replication.factor: -1
    offset.storage.replication.factor: -1
    status.storage.replication.factor: -1
  build:
    output:
      type: docker
      image: image-registry.openshift-image-registry.svc:5000/debezium-example/debezium-connect-mysql:latest
    plugins:
      - name: debezium-mysql-connector
        artifacts:
          - type: tgz
            url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/{debezium-version}/debezium-connector-mysql-{debezium-version}-plugin.tar.gz
EOF

此处我们利用了 OpenShift 内置的镜像仓库，它已作为服务运行在 OpenShift 集群上。

为简单起见，我们跳过了下载构件的校验和验证。如果想确认构件是否已正确下载，可通过 sha512sum 属性指定其校验和。详情请参阅 Strimzi 文档。

如果你已经拥有一个合适的容器镜像（无论是在本地镜像仓库还是远程镜像仓库中，例如 quay.io 或 DockerHub），则可以使用以下简化版本：

$ cat << EOF | oc create -f -
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: debezium-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  version: 3.1.0
  image: 10.110.154.103/debezium-connect-mysql:latest
  replicas: 1
  bootstrapServers: debezium-cluster-kafka-bootstrap:9092
  config:
    config.providers: secrets
    config.providers.secrets.class: io.strimzi.kafka.KubernetesSecretConfigProvider
    group.id: connect-cluster
    offset.storage.topic: connect-cluster-offsets
    config.storage.topic: connect-cluster-configs
    status.storage.topic: connect-cluster-status
    # -1 means it will use the default replication factor configured in the broker
    config.storage.replication.factor: -1
    offset.storage.replication.factor: -1
    status.storage.replication.factor: -1
EOF

您还可以注意到，我们已将 secret provider 配置为使用 Strimzi secret provider。Strimzi secret provider 会为该 Kafka Connect 集群创建服务账号（我们已经将其绑定到相应的角色），并允许 Kafka Connect 访问我们的 Secret 对象。

在创建 Debezium 连接器之前，请检查所有 Pod 是否都已在运行：

openshift pods

创建 Debezium 连接器

要创建 Debezium 连接器，只需创建一个带有相应配置的 KafkaConnector 即可，此处以 MySQL 为例：

$ cat << EOF | oc create -f -
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: debezium-connector-mysql
  labels:
    strimzi.io/cluster: debezium-connect-cluster
spec:
  class: io.debezium.connector.mysql.MySqlConnector
  tasksMax: 1
  config:
    tasks.max: 1
    database.hostname: mysql
    database.port: 3306
    database.user: ${secrets:debezium-example/debezium-secret:username}
    database.password: ${secrets:debezium-example/debezium-secret:password}
    database.server.id: 184054
    topic.prefix: mysql
    database.include.list: inventory
    schema.history.internal.kafka.bootstrap.servers: debezium-cluster-kafka-bootstrap:9092
    schema.history.internal.kafka.topic: schema-changes.inventory
EOF

可以看出，我们在连接器配置中并没有直接使用明文的用户名和密码，而是引用了我们之前创建的 Secret 对象。

验证部署

为了确认一切正常，例如你可以开始监视 mysql.inventory.customers Kafka 主题：

$ oc run -n debezium-example -it --rm --image=quay.io/debezium/tooling:1.2  --restart=Never watcher -- kcat -b debezium-cluster-kafka-bootstrap:9092 -C -o beginning -t mysql.inventory.customers

连接到 MySQL 数据库：

$ oc run -n debezium-example -it --rm --image=mysql:8.2 --restart=Never --env MYSQL_ROOT_PASSWORD=debezium mysqlterm -- mysql -hmysql -P3306 -uroot -pdebezium

对 customers 表做一些更改：

sql> update customers set first_name="Sally Marie" where id=1001;

现在，你应该能够在 Kafka 主题上观察到变更事件：

{
...
  "payload": {
    "before": {
      "id": 1001,
      "first_name": "Sally",
      "last_name": "Thomas",
      "email": "[email protected]"
    },
    "after": {
      "id": 1001,
      "first_name": "Sally Marie",
      "last_name": "Thomas",
      "email": "[email protected]"
    },
    "source": {
      "version": "{debezium-version}",
      "connector": "mysql",
      "name": "mysql",
      "ts_ms": 1646300467000,
      "ts_us": 1646300467000000,
      "ts_ns": 1646300467000000000,
      "snapshot": "false",
      "db": "inventory",
      "sequence": null,
      "table": "customers",
      "server_id": 223344,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 401,
      "row": 0,
      "thread": null,
      "query": null
    },
    "op": "u",
    "ts_ms": 1646300467746,
    "ts_us": 1646300467746178,
    "ts_ns": 1646300467746178963,
    "transaction": null
  }
}

如果你对在 Kubernetes 或 OpenShift 上运行 Debezium 有任何问题或需求，请在我们的用户组或 Debezium开发者聊天室中告诉我们。

评论

登录后参与评论

正在加载评论…