KnowForge/Debezium 简体中文
运营

在 Kubernetes 上运行

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

在 Kubernetes 上部署 Debezium

Debezium 可以轻松部署到开源容器管理平台 Kubernetes 上。该部署利用 Strimzi 项目，该项目旨在通过自定义资源（custom resources）简化 Apache Kafka 在 Kubernetes 上的部署。

要测试你的部署，可以使用 minikube，它会在本地机器上启动一个 Kubernetes 集群。有关如何在机器上安装 minikube 的详细信息，请参阅 minikube 文档。如果你想完全在 minikube 上按照本文档所述测试 Debezium 部署，则需要在 minikube 上搭建一个不安全的容器镜像仓库。为此，你需要使用 --insecure-registry 标志启动 minikube：

$ minikube start --insecure-registry "10.0.0.0/24"

10.0.0.1 是默认的服务集群 IP，因此该配置允许在整个集群内拉取镜像。更多详情请参阅 minikube 文档。你还需要启用 minikube 的 registry 插件：

$ minikube addons enable registry

前提条件

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

$ kubectl create ns debezium-example

部署 Strimzi Operator

如上所述，对于 Debezium 的部署，我们将使用 Strimzi，它负责在 Kubernetes 上管理 Kafka 的部署。有关如何在您的 Kubernetes 集群上部署 Strimzi 的更多详细信息，请参阅 Strimzi 部署文档。

安装 Strimzi 最简单的方式是通过 Operator Lifecycle Manager（OLM）。如果您的集群尚未安装 OLM，可以通过运行以下命令进行安装：

$ curl -sL https://github.com/operator-framework/operator-lifecycle-manager/releases/download/v0.20.0/install.sh | bash -s v0.20.0

现在，安装 Strimzi operator 本身：

$ kubectl create -f https://operatorhub.io/install/strimzi-kafka-operator.yaml

为数据库创建 Secret

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

让我们先创建 Secret 对象：

$ cat << EOF | kubectl create -n debezium-example -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 | kubectl create -n debezium-example -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 | kubectl create -n debezium-example -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

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

部署 Apache Kafka

接下来，部署一个（单节点的）Kafka 集群：

$ cat << EOF | kubectl 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
  • 等待其就绪：
$ kubectl wait kafka/debezium-cluster --for=condition=Ready --timeout=300s -n debezium-example

部署数据源

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

apiVersion: v1
kind: Service
metadata:
  name: mysql
  labels:
    app: mysql
spec:
  ports:
    - port: 3306
  selector:
    app: mysql
$ cat << EOF | kubectl create -n debezium-example -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 | kubectl create -n debezium-example -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: 10.110.154.103/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

你需要将注册表的 IP 地址 10.110.154.103 替换为你能够推送镜像的注册表地址。如果是在启用了 registry 插件的 minikube 上运行，则可以将镜像推送到 minikube 内部的注册表中。注册表的 IP 地址可以通过例如运行以下命令来获取：

$ kubectl -n kube-system get svc registry -o jsonpath='{.spec.clusterIP}'
为简化起见，我们跳过了对下载构件的校验和验证。如果希望确认构件是否已正确下载，可通过 sha512sum 属性指定其校验和。详情请参阅 Strimzi 文档。

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

$ cat << EOF | kubectl create -n debezium-example -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

另请注意，我们已经配置了 Strimzi 密钥提供者。该密钥提供者会为这个 Kafka Connect 集群创建一个服务账号（我们已将其绑定到相应的角色），从而允许 Kafka Connect 访问我们的 Secret 对象。

创建 Debezium 连接器

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

$ cat << EOF | kubectl create -n debezium-example -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 主题，例如：

$ kubectl 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 数据库：

$ kubectl 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": 1646300467746041,
    "ts_ns": 1646300467746041975,
    "transaction": null
  }
}

评论

登录后参与评论

正在加载评论…