使用 Debezium 连接到 Spanner Omni

Debezium 是一个用于变更数据捕获 (CDC) 的开源分布式平台。适用于 Spanner 的 Debezium 连接器可从 Spanner 变更数据流中捕获行级更改，并将其流式传输到 Apache Kafka 主题。

Debezium 连接器与 Spanner Omni 的搭配使用方式与 Spanner 相同。本文档介绍了如何配置 Debezium 连接器以连接到 Spanner Omni。

Spanner Omni Debezium 连接支持三种安全配置：纯文本、TLS 和双向 TLS (mTLS)。

准备工作

如需将 Debezium 连接器与 Spanner Omni 搭配使用，请确保满足以下前提条件：

  • 在 Kafka Connect 插件目录中下载并安装 Debezium 连接器 for Spanner 插件版本 3.7.0 或更高版本。
  • 设置 Apache Kafka 和 Kafka Connect。如需了解详情，请参阅 Debezium 安装指南。
  • 在 Spanner Omni 中创建数据库和变更数据流。如需了解详情，请参阅 Spanner 文档中的变更数据流。

配置 Debezium 连接器

如需将 Debezium 连接器连接到 Spanner Omni，请在 Kafka Connect 连接器配置中指定以下属性：

  • connector.class：设置为 io.debezium.connector.spanner.SpannerConnector。
  • spanner.type：设置为 OMNI。如果设置为 OMNI，gcp.spanner.project.id 和 gcp.spanner.instance.id 会自动设置为 default，并且在配置中不是必需的。
  • gcp.spanner.host：Spanner Omni 端点，例如 http://localhost:15000（用于纯文本）或 https://localhost:15000（用于 TLS 和 mTLS）。
  • gcp.spanner.database.id：Spanner Omni 数据库的 ID。
  • gcp.spanner.change-stream.name：要捕获的变更数据流的名称。
  • spanner.omni.use.plaintext：可选。设置为 true 以建立纯文本连接。
  • spanner.omni.client.cert.path：可选。用于 mTLS 连接的客户端证书文件的路径。
  • spanner.omni.client.key.path：可选。用于 mTLS 连接的 PKCS#8 格式的客户端私钥文件的路径。

建立连接

以下示例展示了如何为每种受支持的安全配置配置 Debezium 连接器：

纯文本

如需建立纯文本连接，请将 spanner.type 设置为 OMNI，使用 http:// 指定端点，并将 spanner.omni.use.plaintext 设置为 true：

{
  "name": "CONNECTOR_NAME",
  "config": {
    "connector.class": "io.debezium.connector.spanner.SpannerConnector",
    "tasks.max": "1",
    "spanner.type": "OMNI",
    "gcp.spanner.host": "http://ENDPOINT",
    "gcp.spanner.database.id": "DATABASE_ID",
    "gcp.spanner.change-stream.name": "CHANGE_STREAM_NAME",
    "spanner.omni.use.plaintext": "true"
  }
}

替换以下内容：

  • CONNECTOR_NAME：Debezium 连接器实例的名称，例如 spanner-omni-connector。

  • ENDPOINT：Spanner Omni 实例的端点，例如 localhost:15000。

  • DATABASE_ID：Spanner Omni 数据库的 ID，例如 test-db。

  • CHANGE_STREAM_NAME：数据库中变更数据流的名称，例如 my_change_stream。

TLS

如需建立 TLS 连接，请按照配置 Java 信任库中的说明将 Spanner Omni CA 证书添加到 Java 信任库。将 spanner.type 设置为 OMNI，并使用 https:// 指定端点：

{
  "name": "CONNECTOR_NAME",
  "config": {
    "connector.class": "io.debezium.connector.spanner.SpannerConnector",
    "tasks.max": "1",
    "spanner.type": "OMNI",
    "gcp.spanner.host": "https://ENDPOINT",
    "gcp.spanner.database.id": "DATABASE_ID",
    "gcp.spanner.change-stream.name": "CHANGE_STREAM_NAME"
  }
}

替换以下内容：

  • CONNECTOR_NAME：Debezium 连接器实例的名称，例如 spanner-omni-connector。

  • ENDPOINT：Spanner Omni 实例的端点，例如 localhost:15000。

  • DATABASE_ID：Spanner Omni 数据库的 ID，例如 test-db。

  • CHANGE_STREAM_NAME：数据库中变更数据流的名称，例如 my_change_stream。

mTLS

如需建立 mTLS 连接，请按照配置 Java 信任库中的说明，将 Spanner Omni CA 证书添加到 Java 信任库。将 spanner.type 设置为 OMNI，使用 https:// 指定端点，并指定客户端证书和客户端私钥的路径。客户端私钥必须采用符合 Java 标准的 PKCS#8 格式，如 Java SDK mTLS 说明中所述：

{
  "name": "CONNECTOR_NAME",
  "config": {
    "connector.class": "io.debezium.connector.spanner.SpannerConnector",
    "tasks.max": "1",
    "spanner.type": "OMNI",
    "gcp.spanner.host": "https://ENDPOINT",
    "gcp.spanner.database.id": "DATABASE_ID",
    "gcp.spanner.change-stream.name": "CHANGE_STREAM_NAME",
    "spanner.omni.client.cert.path": "PATH_TO_CLIENT_CERT",
    "spanner.omni.client.key.path": "PATH_TO_CLIENT_KEY"
  }
}

替换以下内容：

  • CONNECTOR_NAME：Debezium 连接器实例的名称，例如 spanner-omni-connector。

  • ENDPOINT：Spanner Omni 实例的端点，例如 localhost:15000。

  • DATABASE_ID：Spanner Omni 数据库的 ID，例如 test-db。

  • CHANGE_STREAM_NAME：数据库中变更数据流的名称，例如 my_change_stream。

  • PATH_TO_CLIENT_CERT：客户端证书文件的路径。

  • PATH_TO_CLIENT_KEY：客户端私钥文件（采用 PKCS#8 格式）的路径。

后续步骤