Debezium 操作员

概述
使用 Helm 部署 Debezium Operator
跟踪 Operator 部署进度
DebeziumServer 自定义资源
DebeziumServer 规范
完整的 Debezium Server 配置示例

概述

Debezium Operator 是一个 Kubernetes Operator，它在 Kubernetes 和 OpenShift 平台上提供 Debezium Server 实例的自动化部署和管理。该 Operator 实现了标准的 Kubernetes Operator 模式，可以将用户定义的 DebeziumServer 自定义资源转化为运行中的变更数据捕获 (CDC) 流水线，将数据库更改流式传输到各种 sink 系统。

使用 Helm 部署 Debezium Operator

Helm 是安装 Operator 的首选方法。Operator 可在 Debezium Helm Chart 仓库中找到。您可以将 Operator 安装到现有命名空间，也可以安装到一个新的命名空间。

运行以下命令以创建专用命名空间
```
kubectl create namespace debezium
```

添加 Debezium charts 仓库并运行以下命令来安装 Operator

helm repo add debezium https://charts.debezium.io
helm install my-debezium-operator debezium/debezium-operator --version 3.2.0-final -n debezium

您可以在 Debezium Helm Chart 仓库中找到 Operator 的可用版本。

跟踪 Operator 部署进度

您可以运行命令来监视部署进度并查看 Operator 日志。

过程

要监视部署进度，请运行以下命令
```
kubectl get pod -n debezium --watch
```

要监视 Operator 日志，请运行以下命令

kubectl logs deployment/debezium-operator -n debezium -f

Operator 运行后，它会监视新的自定义资源，并根据这些自定义资源创建 Debezium Server 实例。

DebeziumServer 自定义资源

您可以通过在 DebeziumServer 自定义资源中声明属性来指定 Debezium Server 实例的配置。

示例 1. Debezium Server 自定义资源的基本结构

apiVersion: debezium.io/v1alpha1
kind: DebeziumServer
metadata:
  name: my-debezium-server
spec:
  image: String
  version: String
  runtime:
    environment:
      vars:
        - name: String
          value: String
      from:  # EnvFromSource array
        - sourceName: String
          # add other EnvFromSource fields as needed
    storage:
      data:
        type: persistent | ephemeral
        claimName: String  # required if type is "persistent"
      external:  # Volume array
        - name: String
          # add other Volume fields as needed
    jmx:
      enabled: boolean
      port: int  # defaults to 1099
    templates:
      container:
        resources: ResourceRequirements
        securityContext: SecurityContext
      pod:
        metadata:
          annotations:  # Map<String, String>
            key: value
          labels:  # Map<String, String>
            key: value
        imagePullSecrets:  # List
          - name: String
        affinity: Affinity
        securityContext: PodSecurityContext
  quarkus:
    config:
      # quarkus properties
      format:
        value:
          type: String
          config:
            # format properties
      key:
        type: String
        config:
          # format properties
      header:
        type: String
        config:
          # format properties
  transforms:
    - type: String
      predicate: String
      negate: Boolean
      config:
        # transformation properties
  predicates:
    name:
      type: String
      config:
        # predicate properties
  sink:
    type: String
    config:
      # sink properties
  source:
    class: String
    config:
      # source connector properties

DebeziumServer 规范

DebeziumServer Schema 参考

表 1. DebeziumServer 属性
属性	Type	Default (默认值)	描述
`spec`	`DebeziumServerSpec`	无默认值	Debezium Server 的规范
`status`	`DebeziumServerStatus`	无默认值	Debezium Server 实例的状态。

DebeziumServerStatus Schema 参考

在以下位置使用: DebeziumServer

表 2. DebeziumServerStatus 属性
属性	Type	Default (默认值)	描述
`conditions`	`List<Condition>`	无默认值	状态条件列表
`observedGeneration`	Long	0	最新的观察到的代

Condition Schema 参考

在以下位置使用: DebeziumServerStatus

表 3. Condition 属性
属性	Type	Default (默认值)	描述
`status`	String	无默认值	条件的状�?�，可以是 True、False 或 Unknown。
`message`	String	无默认值	人类可读的消息，指示有关条件最后一次转换的详细信息。
`type`	String	无默认值	条件的唯一标识符。

DebeziumServerSpec Schema 参考

在以下位置使用: DebeziumServer

表 4. DebeziumServerSpec 属性
属性	Type	Default (默认值)	描述
`image`	String	无默认值	用于 Debezium Server 容器的镜像。此属性优先于版本。
`version`	String	与 Operator 相同	要使用的 Debezium Server 的版本。
`sink`	`Sink`	无默认值	Sink 配置。
`source`	`源`	无默认值	Debezium 源连接器配置。
`format`	`Format`	无默认值	消息输出格式配置。
`quarkus`	`Quarkus`	无默认值	传递给 Debezium Server 进程的 Quarkus 配置。
`runtime`	`Runtime`	无默认值	指定您是否可以修改 Debezium Server 运行时的各种方面。
`transforms`	`List<Transformation>`	无默认值	此 Debezium Server 实例使用的单消息转换。
`predicates`	`Map<String, Predicate>`	无默认值	此 Debezium Server 实例使用的谓词。

Sink Schema 参考

在以下位置使用: DebeziumServerSpec

表 5. Sink 属性
属性	Type	Default (默认值)	描述
`type`	String	无默认值	此 Debezium Server 实例识别的 Sink 类型。
`config`	Map	无默认值	Sink 配置属性。

Source Schema 参考

在以下位置使用: DebeziumServerSpec

表 6. Source 属性
属性	Type	Default (默认值)	描述
`sourceClass`	String	无默认值	源连接器 Java 类的完全限定名。
`offset`	`Offset`	无默认值	Offset 存储配置
`schemaHistory`	`SchemaHistory`	无默认值	Schema history 存储配置
`config`	Map	无默认值	Source 连接器配置属性。

Format Schema 参考

在以下位置使用: DebeziumServerSpec

表 7. Format 属性
属性	Type	Default (默认值)	描述
`key`	`FormatType`	无默认值	消息键格式配置。
`value`	`FormatType`	无默认值	消息值格式配置。
`header`	`FormatType`	无默认值	消息头格式配置。

FormatType Schema 参考

在以下位置使用: Format

表 8. FormatType 属性
属性	Type	Default (默认值)	描述
`type`	String	json	Debezium Server 识别的格式类型。
`config`	Map	无默认值	格式配置属性。

Quarkus Schema 参考

在以下位置使用: DebeziumServerSpec

表 9. Quarkus 属性
属性	Type	Default (默认值)	描述
`config`	Map	无默认值	Quarkus 配置属性。

Runtime Schema 参考

在以下位置使用: DebeziumServerSpec

表 10. Runtime 属性
属性	Type	Default (默认值)	描述
`api`	`RuntimeApi`	无默认值	API 配置
`storage`	`RuntimeStorage`	无默认值	存储配置
`environment`	`RuntimeEnvironment`	无默认值	此 Debezium Server 实例使用的其他环境变量。
`jmx`	`JmxConfig`	无默认值	JMX 配置。
`templates`	`Templates`	无默认值	Debezium Server 资源模板。
`serviceAccount`	String	无默认值	用于运行 Debezium Server Pod 的现有服务帐户
`metrics`	`Metrics`	无默认值	Metrics 配置

RuntimeApi Schema 参考

在以下位置使用: Runtime

表 11. RuntimeApi 属性
属性	Type	Default (默认值)	描述
`enabled`	boolean	无默认值	是否为该 Debezium Server 实例启用 API
`port`	int	8080	用于公开 API 的 k8s 服务的端口号

RuntimeStorage Schema 参考

在以下位置使用: Runtime

表 12. RuntimeStorage 属性
属性	Type	Default (默认值)	描述
`data`	`DataStorage`	无默认值	此 Debezium Server 实例使用的文件存储配置。
`external`	`List<Volume>`	无默认值	挂载到 /debezium/external 的附加卷

RuntimeEnvironment Schema 参考

在以下位置使用: Runtime

表 13. RuntimeEnvironment 属性
属性	Type	Default (默认值)	描述
`vars`	`List<ContainerEnvVar>`	无默认值	应用于容器的环境变量。
`from`	`List<EnvFromSource>`	无默认值	从容器中的 ConfigMaps 或 Secrets 设置的附加环境变量。

JmxConfig Schema 参考

在以下位置使用: Runtime

表 14. JmxConfig 属性
属性	Type	Default (默认值)	描述
`enabled`	boolean	false	是否为该 Debezium Server 实例启用 JMX。
`port`	int	1099	JMX 端口。
`authentication`	`JmxAuthentication`	无默认值	JMX 认证配置。

JmxAuthentication Schema 参考

在以下位置使用: JmxConfig

表 15. JmxAuthentication 属性
属性	Type	Default (默认值)	描述
`enabled`	boolean	false	是否为该 Debezium Server 实例启用 JMX 认证。
`secret`	String	无默认值	提供凭据文件的 Secret
`accessFile`	String	jmxremote.access	JMX 访问文件名和 Secret 密钥
`passwordFile`	String	jmxremote.password	JMX 密码文件名和 Secret 密钥

Templates Schema 参考

在以下位置使用: Runtime

表 16. Templates 属性
属性	Type	Default (默认值)	描述
`container`	`ContainerTemplate`	无默认值	容器模板
`pod`	`PodTemplate`	无默认值	Pod 模板。
`volumeClaim`	`PersistentVolumeClaimSpec`	无默认值	如果未指定显式声明，则用于数据卷的 PVC 模板。

Metrics Schema 参考

在以下位置使用: Runtime

表 17. Metrics 属性
属性	Type	Default (默认值)	描述
`jmxExporter`	`JmxExporter`	无默认值	Prometheus JMX exporter 配置

JmxExporter Schema 参考

在以下位置使用: Metrics

表 18. JmxExporter 属性
属性	Type	Default (默认值)	描述
`enabled`	boolean	无默认值	启用 JMX Prometheus exporter
`configFrom`	ConfigMapKeySelector	无默认值	Config map 键的引用，其值将用作配置文件

Transformation Schema 参考

在以下位置使用: DebeziumServerSpec

表 19. Transformation 属性
属性	Type	Default (默认值)	描述
`type`	String	无默认值	实现转换的 Java 类的完全限定名。
`config`	Map	无默认值	转换特定的配置属性。
`predicate`	String	无默认值	将应用于此转换的谓词的名称。
`negate`	boolean	false	确定应用的谓词结果是否将被否定。

Predicate Schema 参考

在以下位置使用: DebeziumServerSpec

表 20. Predicate 属性
属性	Type	Default (默认值)	描述
`type`	String	无默认值	实现谓词的 Java 类的完全限定名。
`config`	Map	无默认值	谓词配置属性。

ConfigMapOffsetStore Schema 参考

在以下位置使用: Offset

表 21. ConfigMapOffsetStore 属性
属性	Type	Default (默认值)	描述
`name`	String	无默认值	Offset ConfigMap 的名称
`config`	Map	无默认值	其他存储配置属性。

ContainerEnvVar Schema 参考

在以下位置使用: RuntimeEnvironment

表 22. ContainerEnvVar 属性
属性	Type	Default (默认值)	描述
`name`	String	无默认值	环境变量的名称。
`value`	String	无默认值	环境变量的值。

ContainerTemplate Schema 参考

在以下位置使用: Templates

表 23. ContainerTemplate 属性
属性	Type	Default (默认值)	描述
`resources`	`ResourceRequirements`	无默认值	CPU 和内存资源要求。
`securityContext`	`SecurityContext`	无默认值	容器安全上下文。
`probes`	`Probes`	无默认值	容器探测配置。

CustomStore Schema 参考

在以下位置使用: Offset, SchemaHistory

表 24. CustomStore 属性
属性	Type	Default (默认值)	描述
`type`	String	无默认值	实现存储的 Java 类的完全限定名。
`config`	Map	无默认值	存储配置属性。

DataStorage Schema 参考

在以下位置使用: RuntimeStorage

表 25. DataStorage 属性
属性	Type	Default (默认值)	描述
`type`	ephemeral,persistent	ephemeral	存储类型。
`claimName`	String	无默认值	持久存储的持久卷声明名称。

FileOffsetStore Schema 参考

在以下位置使用: Offset

表 26. FileOffsetStore 属性
属性	Type	Default (默认值)	描述
`fileName`	String	无默认值	Offset 文件的名称（相对于数据根目录）
`config`	Map	无默认值	其他存储配置属性。

FileSchemaHistoryStore Schema 参考

在以下位置使用: SchemaHistory

表 27. FileSchemaHistoryStore 属性
属性	Type	Default (默认值)	描述
`fileName`	String	无默认值	Offset 文件的名称（相对于数据根目录）
`config`	Map	无默认值	其他存储配置属性。

InMemoryOffsetStore Schema 参考

在以下位置使用: Offset

表 28. InMemoryOffsetStore 属性
属性	Type	Default (默认值)	描述
`config`	Map	无默认值	其他存储配置属性。

InMemorySchemaHistoryStore Schema 参考

在以下位置使用: SchemaHistory

表 29. InMemorySchemaHistoryStore 属性
属性	Type	Default (默认值)	描述
`config`	Map	无默认值	其他存储配置属性。

JdbcOffsetStore Schema 参考

在以下位置使用: Offset

表 30. JdbcOffsetStore 属性
属性	Type	Default (默认值)	描述
`table`	`JdbcOffsetTableConfig`	无默认值	Offset 表的配置
`url`	String	无默认值	JDBC 连接 URL
`user`	String	无默认值	连接存储数据库使用的用户名
`password`	String	无默认值	连接存储数据库使用的密码
`retryDelay`	long	无默认值	连接失败时的重试延迟（以毫秒为单位）
`maxRetries`	int	无默认值	连接失败时的最大重试次数
`config`	Map	无默认值	其他存储配置属性。

JdbcOffsetTableConfig Schema 参考

在以下位置使用: JdbcOffsetStore

表 31. JdbcOffsetTableConfig 属性
属性	Type	Default (默认值)	描述
`name`	String	无默认值	Offset 表的名称
`ddl`	String	无默认值	用于创建 Offset 表的 DDL 语句
`select`	String	无默认值	用于从 Offset 表中选择的语句
`insert`	String	无默认值	用于向 Offset 表插入的语句
`delete`	String	无默认值	用于更新 Offset 表的语句

JdbcSchemaHistoryStore Schema 参考

在以下位置使用: SchemaHistory

表 32. JdbcSchemaHistoryStore 属性
属性	Type	Default (默认值)	描述
`table`	`JdbcSchemaHistoryTableConfig`	无默认值	Offset 表的配置
`url`	String	无默认值	JDBC 连接 URL
`user`	String	无默认值	连接存储数据库使用的用户名
`password`	String	无默认值	连接存储数据库使用的密码
`retryDelay`	long	无默认值	连接失败时的重试延迟（以毫秒为单位）
`maxRetries`	int	无默认值	连接失败时的最大重试次数
`config`	Map	无默认值	其他存储配置属性。

JdbcSchemaHistoryTableConfig Schema 参考

在以下位置使用: JdbcSchemaHistoryStore

表 33. JdbcSchemaHistoryTableConfig 属性
属性	Type	Default (默认值)	描述
`name`	String	无默认值	Offset 表的名称
`ddl`	String	无默认值	用于创建 Schema History 表的 DDL 语句
`select`	String	无默认值	用于从 Schema History 表中选择的语句
`insert`	String	无默认值	用于向 Schema History 表插入的语句
`dataExistsSelect`	String	无默认值	用于检查 Schema History 表中是否存在某些数据的语句

KafkaOffsetStore Schema 参考

在以下位置使用: Offset

表 34. KafkaOffsetStore 属性
属性	Type	Default (默认值)	描述
`props`	Map	无默认值	其他 Kafka 客户端属性。
`bootstrapServers`	String	无默认值	连接器用于建立与 Kafka 集群的初始连接的主机/端口对列表
`topic`	String	无默认值	用于存储 Offset 的 Kafka Topic 名称
`partitions`	int	无默认值	创建 Offset 存储 Topic 时使用的分区数
`replicationFactor`	int	无默认值	创建 Offset 存储 Topic 时使用的副本因子
`config`	Map	无默认值	其他存储配置属性。

KafkaSchemaHistoryStore Schema 参考

在以下位置使用: SchemaHistory

表 35. KafkaSchemaHistoryStore 属性
属性	Type	Default (默认值)	描述
`bootstrapServers`	String	无默认值	连接器用于建立与 Kafka 集群的初始连接的主机/端口对列表
`topic`	String	无默认值	用于存储 Offset 的 Kafka Topic 名称
`partitions`	int	无默认值	创建 Offset 存储 Topic 时使用的分区数
`replicationFactor`	int	无默认值	创建 Offset 存储 Topic 时使用的副本因子
`config`	Map	无默认值	其他存储配置属性。

MetadataTemplate Schema 参考

在以下位置使用: PodTemplate

表 36. MetadataTemplate 属性
属性	Type	Default (默认值)	描述
`labels`	Map<String, String>	无默认值	添加到 Kubernetes 资源的标签
`annotations`	Map<String, String>	无默认值	添加到 Kubernetes 资源的注释

Offset Schema 参考

在以下位置使用: Source

表 37. Offset 属性
属性	Type	Default (默认值)	描述
`file`	`FileOffsetStore`	无默认值	文件后备 Offset 存储配置
`memory`	`InMemoryOffsetStore`	无默认值	内存后备 Offset 存储配置
`redis`	`RedisOffsetStore`	无默认值	Redis 后备 Offset 存储配置
`kafka`	`KafkaOffsetStore`	无默认值	Kafka 后备存储配置
`jdbc`	`JdbcOffsetStore`	无默认值	JDBC 后备存储配置
`configMap`	`ConfigMapOffsetStore`	无默认值	Config Map 后备 Offset 存储配置
`store`	`CustomStore`	无默认值	任意 Offset 存储配置
`flushMs`	long	60000	尝试提交 Offset 的间隔

PodTemplate Schema 参考

在以下位置使用: Templates

表 38. PodTemplate 属性
属性	Type	Default (默认值)	描述
`metadata`	`MetadataTemplate`	无默认值	应用于资源的元数据。
`imagePullSecrets`	`List<LocalObjectReference>`	无默认值	用于拉取此 Pod 使用的任何镜像的 Secret 的本地引用列表。
`affinity`	`Affinity`	无默认值	Pod 亲和性规则
`securityContext`	`PodSecurityContext`	无默认值	Pod 级别的安全属性和容器设置

Probe Schema 参考

在以下位置使用: Probes

表 39. Probe 属性
属性	Type	Default (默认值)	描述
`initialDelaySeconds`	int	5	容器启动后多长时间（秒）开始执行探测。
`periodSeconds`	int	10	每次执行探测的间隔（秒）。
`timeoutSeconds`	int	10	探测超时之前的秒数。
`failureThreshold`	int	3	连续失败次数，在此之后整个检查将失败。

Probes Schema 参考

在以下位置使用: ContainerTemplate

表 40. Probes 属性
属性	Type	Default (默认值)	描述
`readiness`	`Probe`	无默认值	应用于容器的就绪探测配置。
`liveness`	`Probe`	无默认值	应用于容器的存活探测配置。

RedisOffsetStore Schema 参考

在以下位置使用: Offset

表 41. RedisOffsetStore 属性
属性	Type	Default (默认值)	描述
`address`	String	无默认值	用于连接的 Redis 主机:端口
`user`	String	无默认值	Redis 用户名
`password`	String	无默认值	Redis 密码
`sslEnabled`	boolean	false	Redis 用户名
`key`	String	无默认值	Redis 哈希键
`wait`	`RedisStoreWaitConfig`	无默认值	配置副本写入的验证
`config`	Map	无默认值	其他存储配置属性。

RedisSchemaHistoryStore Schema 参考

在以下位置使用: SchemaHistory

表 42. RedisSchemaHistoryStore 属性
属性	Type	Default (默认值)	描述
`address`	String	无默认值	用于连接的 Redis 主机:端口
`user`	String	无默认值	Redis 用户名
`password`	String	无默认值	Redis 密码
`sslEnabled`	boolean	false	Redis 用户名
`key`	String	无默认值	Redis 哈希键
`wait`	`RedisStoreWaitConfig`	无默认值	配置副本写入的验证
`config`	Map	无默认值	其他存储配置属性。

RedisStoreWaitConfig Schema 参考

在以下位置使用: RedisOffsetStore, RedisSchemaHistoryStore

表 43. RedisStoreWaitConfig 属性
属性	Type	Default (默认值)	描述
`enabled`	boolean	false	如果 Redis 具有副本，这允许验证数据是否已写入副本
`timeoutMs`	long	1000	等待副本的超时时间（毫秒）
`retry`	boolean	false	等待副本时启用重试
`retryDelayMs`	long	1000	等待时重试的延迟

SchemaHistory Schema 参考

在以下位置使用: Source

表 44. SchemaHistory 属性
属性	Type	Default (默认值)	描述
`file`	`FileSchemaHistoryStore`	无默认值	文件后备 Schema history 存储配置
`memory`	`InMemorySchemaHistoryStore`	无默认值	内存后备 Schema history 存储配置
`redis`	`RedisSchemaHistoryStore`	无默认值	Redis 后备 Schema history 存储配置
`kafka`	`KafkaSchemaHistoryStore`	无默认值	Kafka 后备 Schema history 存储配置
`jdbc`	`JdbcSchemaHistoryStore`	无默认值	JDBC 后备 Schema history 存储配置
`store`	`CustomStore`	无默认值	任意 Schema history 存储配置
`config`	Map	无默认值	其他通用 Schema history 存储配置属性。

完整的 Debezium Server 配置示例

以下示例显示了一个完整的 Debezium Server 自定义资源，其中包含前面表格中的属性。

示例 2. 完整的 Debezium Server 自定义资源

apiVersion: debezium.io/v1alpha1
kind: DebeziumServer
metadata:
  name: my-debezium-server
spec:
  version: "3.2.0"
  source:
    class: io.debezium.connector.mysql.MySqlConnector
    config:
      database.hostname: mysql-server
      database.port: 3306
      database.user: debezium
      database.password: secret
      database.server.id: 184054
      topic.prefix: mysql
      database.include.list: inventory
    offset:
      type: kafka
      config:
        bootstrap.servers: kafka:9092
        topic: debezium-offsets
    schemaHistory:
      type: kafka
      config:
        bootstrap.servers: kafka:9092
        topic: debezium-schema-history
  sink:
    type: kafka
    config:
      bootstrap.servers: kafka:9092
  format:
    key:
      type: json
    value:
      type: json
      config:
        schemas.enable: false
  runtime:
    jmx:
      enabled: true
      port: 1099
    storage:
      data:
        type: persistent
        claimName: debezium-data-pvc
    metrics:
      jmxExporter:
        enabled: true
  transforms:
    - type: io.debezium.transforms.UnwrapFromEnvelope
      config:
        drop.tombstones: false
      predicate: inventory-filter
      negate: false
  predicates:
    inventory-filter:
      type: org.apache.kafka.connect.transforms.predicates.TopicNameMatches
      config:
        pattern: "mysql.inventory.*"
  quarkus:
    config:
      quarkus.log.level: INFO
      quarkus.http.port: 8080