Debezium系列之:在 Kubernetes 上部署 Debezium
Debezium 可以轻松部署在开源容器管理平台 Kubernetes 上。该部署利用了 Strimzi 项目,该项目旨在通过自定义资源简化 Kubernetes 上 Apache Kafka 的部署。为了测试您的部署,您可以使用 minikube,它会在本地计算机上启动 Kubernetes 集群。如果您想在 minikube 上完全测试本文档中描述的 Debezium 部署,则需要在 mini
这Debezium系列之:在 Kubernetes 上部署 Debezium
K8s相关知识可以阅读博主以下几篇技术博客:
- K8s系列之:搭建高可用K8s v1.23.5集群详细步骤,3个master节点,3个Node节点
- K8s系列之:Pod的基本用法
- k8s系列之:kubectl子命令详解一
- k8s系列之:kubectl子命令详解二
- 更多K8s知识点详见博主K8s系列文章
更多Debezium内容请阅读博主Debezium专栏,博主会持续更新Debezium专栏:
一、概述
Debezium 可以轻松部署在开源容器管理平台 Kubernetes 上。该部署利用了 Strimzi 项目,该项目旨在通过自定义资源简化 Kubernetes 上 Apache Kafka 的部署。
为了测试您的部署,您可以使用 minikube,它会在本地计算机上启动 Kubernetes 集群。如果您想在 minikube 上完全测试本文档中描述的 Debezium 部署,则需要在 minikube 上设置不安全的容器映像注册表。为此,您需要使用 --insecure-registry 标志启动 minikube:
$ minikube start --insecure-registry "10.0.0.0/24"
10.0.0.1 是默认的服务集群 IP,因此此设置允许在整个集群内拉取镜像。您还需要启用注册表 minikube 插件:
minikube addons enable registry
二、先决条件
为了使容器与集群上的其他工作负载分开,请为 Debezium 创建专用命名空间。在本文档的其余部分中,将使用 debezium-example 命名空间:
kubectl create ns debezium-example
部署 Strimzi Operator
如上所述,对于 Debezium 部署,我们将使用 Strimzi,它管理 Kubernetes 上的 Kafka 部署。
部署minikube详细步骤可以参考博主下面这篇技术博客:
安装 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
三、为数据库创建Secrets
稍后,在部署 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
用户名和密码包含用于连接 MySQL 数据库的 Base64 编码凭据 (debezium/dbz),我们稍后将部署该数据库。
现在,我们可以创建一个角色,它引用上一步中创建的秘密:
$ 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 作为 subject.name。
四、部署Apache Kafka
接下来,部署一个(单节点)Kafka 集群:
$ cat << EOF | kubectl create -n debezium-example -f -
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
name: debezium-cluster
spec:
kafka:
replicas: 1
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
storage:
type: jbod
volumes:
- id: 0
type: persistent-claim
size: 100Gi
deleteClaim: false
config:
offsets.topic.replication.factor: 1
transaction.state.log.replication.factor: 1
transaction.state.log.min.isr: 1
default.replication.factor: 1
min.insync.replicas: 1
zookeeper:
replicas: 1
storage:
type: persistent-claim
size: 100Gi
deleteClaim: false
entityOperator:
topicOperator: {}
userOperator: {}
EOF
等待它准备好:
$ kubectl wait kafka/debezium-cluster --for=condition=Ready --timeout=300s -n debezium-example
五、部署数据源
下面将使用MySQL作为数据源。除了使用 MySQL 运行 pod 之外,还需要一个适当的服务来指向带有 DB 本身的 pod。它可以被创建,例如如下:
$ 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:2.3
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连接器为例。
$ 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
您必须将registry的IP地址10.110.154.103替换为可以推送镜像的registry。如果您使用注册表插件在 minikube 上运行它,您可以将映像推送到内部 minikube 注册表中。注册表的 IP 地址可以通过以下方式获得:通过运行
kubectl -n kube-system get svc registry -o jsonpath='{.spec.clusterIP}'
为简单起见,我们跳过了下载工件的校验和验证。如果您想确保工件已正确下载,请通过 sha512sum 属性指定其校验和。
如果您在本地或远程注册表(例如 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 Topic:
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.0 --restart=Never
--env MYSQL_ROOT_PASSWORD=debezium mysqlterm
-- mysql -hmysql -P3306 -uroot -pdebezium
在客户表中做一些更改:
sql> update customers set first_name="Sally Marie" where id=1001;
您现在应该能够观察 Kafka 主题上的更改事件:
{
...
"payload": {
"before": {
"id": 1001,
"first_name": "Sally",
"last_name": "Thomas",
"email": "sally.thomas@acme.com"
},
"after": {
"id": 1001,
"first_name": "Sally Marie",
"last_name": "Thomas",
"email": "sally.thomas@acme.com"
},
"source": {
"version": "{debezium-version}",
"connector": "mysql",
"name": "mysql",
"ts_ms": 1646300467000,
"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,
"transaction": null
}
}
更多推荐
所有评论(0)