存储 Debezium 连接器的状态

概述

Debezium 连接器需要持久化存储，以便在重启之间保留其状态。所有连接器都需要一种机制来为偏移量（offset）提供持久化存储。此外，Db2、MySQL、Oracle 和 SQL Server 等连接器还需要额外的存储空间，用于保存其所谓的内部模式历史（internal schema history），以记录数据库中表结构的变化。

在 Kafka Connect 运行时内部部署时，偏移量存储由以下机制之一自动提供：

如果你在 Debezium Engine 或 Debezium Server 中运行连接器，则必须显式配置偏移量存储。对于基于模式（schema）的数据库连接器，你需要通过设置连接器属性来配置内部模式历史存储。

Kafka

Debezium 可以使用 Kafka 来存储其状态，包括源端偏移量和模式历史。连接器实现 KafkaOffsetBackingStore，将偏移量存储在 Kafka 主题（topic）中（例如 connect-offsets）。这些偏移量确保连接器重启后能够从正确的位置继续读取。连接器将其模式历史存储在一个单独的压缩主题中，例如 schema-changes.inventory。

偏移量存储

内部模式历史存储

文件

可以将连接器的位置（偏移量）持久化到磁盘上的本地文件中。这些偏移量可确保连接器重启后，Debezium 能从上次读取的位置继续读取。文件存储提供了一种简单、快速的偏移量存储机制，非常适合单节点应用或测试场景。

偏移量存储

内部模式历史存储

内存

MemoryOffsetBackingStore 是 Debezium Embedded 用于跟踪源偏移量的易失性内存存储。将偏移量存储在内存中，只会在应用程序运行期间保留偏移状态。如果连接器关闭或崩溃，偏移记录将会丢失。内存存储适用于测试或短生命周期的任务，但不适用于需要持久化偏移量的生产环境。

偏移存储

内部 schema 历史存储

JDBC

该存储使用任意关系数据库来存储偏移数据。你必须为该数据库提供 JDBC 驱动程序。Debezium 可以将数据存储在它捕获事件的同一源数据库中，也可以配置为使用不同的数据库。

Debezium 提供了预先配置好的 DML 和 DDL 语句。你可以使用这些默认语句，也可以用自己的语句覆盖默认语句，以便与各种数据库方言保持兼容，或针对特定用例进行定制。

偏移存储

3.2 之前的已弃用配置

偏移表默认值

创建查询

CREATE TABLE %s (
id VARCHAR(36)      NOT NULL,
offset_key          VARCHAR(1255),
offset_val          VARCHAR(1255),
record_insert_ts    TIMESTAMP NOT NULL,
record_insert_seq   INTEGER NOT NULL),
PRIMARY KEY (id)

Select 查询

SELECT id, offset_key, offset_val FROM %s ORDER BY record_insert_ts, record_insert_seq

Debezium插入查询

INSERT INTO %s(id, offset_key, offset_val, record_insert_ts, record_insert_seq)
    VALUES ( ?, ?, ?, ?, ? )

删除查询

DELETE FROM %s

内置架构历史存储

3.2 之前的弃用配置

历史表默认值

创建查询

CREATE TABLE %s (
    id VARCHAR(36) NOT NULL,
    history_data VARCHAR(65000),
    history_data_seq INTEGER,
    record_insert_ts TIMESTAMP NOT NULL,
    record_insert_seq INTEGER NOT NULL,
    PRIMARY KEY (id, history_data_seq)
)

选择查询

SELECT id, history_data FROM %s
    ORDER BY record_insert_ts, record_insert_seq, id, history_data_seq

数据存在性查询

SELECT * FROM %s LIMIT 1

插入查询

INSERT INTO %s(id, history_data, history_data_seq, record_insert_ts, record_insert_seq) VALUES ( ?, ?, ?, ?, ? )

Redis

Debezium 可以使用 Jedis 客户端 将数据存储在 Redis 缓存中。

Debezium 既可以使用单个 Redis 实例，也可以使用 Redis 集群模式：

单实例模式

连接到单个 Redis 服务器实例。

集群模式

连接到 Redis 集群，以实现高可用性和水平扩展。

要启用 Redis 集群模式，请将 redis.cluster.enabled 属性设置为 true，并在 redis.address 属性中提供以逗号分隔的 host:port 地址。

偏移量存储

内部模式历史存储

Redis 配置示例

单实例模式

# Offset storage configuration
offset.storage=io.debezium.storage.redis.offset.RedisOffsetBackingStore
offset.storage.redis.address=localhost:6379
offset.storage.redis.password=password
offset.storage.redis.cluster.enabled=false

# Schema history storage configuration
schema.history.internal=io.debezium.storage.redis.history.RedisSchemaHistory
schema.history.internal.storage.redis.address=localhost:6379
schema.history.internal.storage.redis.password=password
schema.history.internal.storage.redis.cluster.enabled=false

集群模式

# Offset storage configuration
offset.storage=io.debezium.storage.redis.offset.RedisOffsetBackingStore
offset.storage.redis.address=redis-node-1:7001,redis-node-2:7002,redis-node-3:7003
offset.storage.redis.password=password
offset.storage.redis.cluster.enabled=true

# Schema history storage configuration
schema.history.internal=io.debezium.storage.redis.history.RedisSchemaHistory
schema.history.internal.storage.redis.address=redis-node-1:7001,redis-node-2:7002,redis-node-3:7003
schema.history.internal.storage.redis.password=password
schema.history.internal.storage.redis.cluster.enabled=true

Amazon S3

Debezium 可以使用 Amazon S3 对象存储服务。通常，当你使用 Amazon Managed Streaming for Apache Kafka (Amazon MSK) 部署 Debezium 时，会采用 S3 存储。

内部模式历史记录存储

属性 默认值 说明

schema.history.internal

无默认值

必须设置为 io.debezium.storage.s3.history.S3SchemaHistory。

schema.history.internal.s3.access.key.id

无默认值

（可选）Debezium 用于向 S3 进行身份验证的静态访问密钥标识符。

schema.history.internal.s3.secret.access.key

无默认值

（可选）Debezium 用于向 S3 进行身份验证的 Amazon Web Services (AWS) 秘密密钥。

schema.history.internal.s3.region.name

无默认值

（可选）指定托管 S3 存储桶的区域名称。

schema.history.internal.s3.bucket.name

无默认值

指定存储模式历史记录的 S3 存储桶名称。

schema.history.internal.s3.object.name

无默认值

指定存储模式历史记录的对象在存储桶中的名称。

schema.history.internal.s3.endpoint

无默认值

（可选）指定 Debezium 用于访问 S3 服务的自定义 URL。
请按以下格式提供 URL：http://<server>:<port>;

Azure Blob Storage

Debezium 可以使用 Azure Blob 存储服务来保存数据。通常，当你在 Apache Kafka in Azure HDInsight 服务中部署 Debezium 时，会采用 Azure Blob 存储。

内部模式历史记录存储

RocketMQ

Debezium 可以使用 RocketMqSchemaHistory 类在 Apache RocketMQ 中存储和检索数据库模式变更。

内部模式历史存储

Chronicle Queue

Debezium 连接器在将变更事件传递给运行时框架之前，会先将其缓存在内部队列中。默认情况下，Debezium 使用内存中的有界队列。你可以通过设置 queue.provider.type 连接器属性，将默认队列替换为基于 Chronicle Queue 的实现。Chronicle Queue 将事件写入磁盘上的内存映射文件，从而在高吞吐量期间降低堆内存压力。

目前提供两种 Chronicle Queue 实现：

chronicle

将所有变更事件溢写到磁盘。这样可以将事件完全从 JVM 堆中移除，适用于预期持续高吞吐量并希望最大限度减少垃圾回收开销的场景。

hybrid_chronicle

以内存队列作为主要缓冲区，当缓冲区达到容量上限时将溢出数据写入磁盘。在低流量场景下，不会发生序列化或磁盘 I/O。

JVM 设置

使用 debezium/server 或 debezium/connect 容器镜像时，只需将环境变量 ENABLE_CHRONICLE_QUEUE 设置为 true，所有必需的 JVM 配置都会自动为你完成。

使用其他部署方式时，必须通过配置显式授予 Chronicle Queue 访问若干 JVM 内部 API 的权限。这需要指定以下 JVM 启动参数，这些参数授予 Chronicle Queue 反射访问 Java 核心内部类的权限，以支持其堆外内存管理和低延迟操作所需的机制。

--add-opens=java.base/sun.nio.ch=ALL-UNNAMED
--add-opens=java.base/java.lang=ALL-UNNAMED
--add-opens=java.base/java.lang.reflect=ALL-UNNAMED
--add-opens=java.base/java.io=ALL-UNNAMED
--add-opens=java.base/java.util=ALL-UNNAMED

Chronicle Queue 提供程序

要使用 chronicle 提供程序，请将 queue.provider.type 设置为 chronicle。

混合 Chronicle Queue 提供程序

要使用 hybrid_chronicle 提供程序，请将 queue.provider.type 设置为 hybrid_chronicle。

内存缓冲区保存由 max.queue.size 定义的容量范围内的最新事件。当缓冲区达到容量上限时，在添加新事件之前，最旧的事件会被逐出到磁盘。在轮询时，Debezium 首先从 Chronicle Queue 中取出已逐出的事件，然后从内存缓冲区中取出事件，从而保持严格的 FIFO 顺序。