Snowflake Connector for Kafka

Snowflake Connector for Kafka (v4) 是一种接收器连接器，用于从一个或多个 Apache Kafka 主题读取数据并将这些数据加载到 Snowflake 表中。该连接器基于 Snowflake 的 高性能 Snowpipe Streaming 架构 构建，最高可实现 10 GB/s 的吞吐量，端到端延迟为 5 到 10 秒，同时支持“精确一次”和有序传递语义。

有关 Kafka Connect 及其框架的更多信息，请参阅 Apache Kafka 与 Kafka Connect 框架。

优势

Snowflake Connector for Kafka 利用了 Snowflake 的 高性能 Snowpipe Streaming 架构，该架构专为数据密集型的现代组织打造，能够实现近实时见解。这种下一代架构显著提高了实时引入 Snowflake 的吞吐量、效率和灵活性。

高性能架构具有以下几个关键优势：

  • 卓越的吞吐量和延迟：旨在支持每个表高达 10 GB/秒的引入速度，端到端引入到查询的延迟在 5 到 10 秒内，从而实现近乎实时的分析。

  • 基于吞吐量的统一定价：定价基于引入的数据量 (GB)，与 Snowpipe Streaming 高性能架构采用相同的计费模式。有关定价详细信息，请参阅 Snowpipe Streaming 成本。

  • 增强的性能：使用基于 Rust 的客户端核心，与之前的实现相比，它提供了更高的客户端性能和更低的资源占用。

  • 传输中的数据转换：支持在引入过程中使用 PIPE 对象中的 COPY 命令语法进行数据清洗和整形，允许您在数据到达目标表之前对其进行转换。

  • 服务器端架构验证：通过 PIPE 对象将架构验证从客户端移动到服务器端，确保了数据质量并降低了客户端的复杂性。无效记录会被捕获到 错误表 中，以便检查和重放。

  • 预聚类功能：当目标表定义了聚类键时，可以在数据引入期间对数据进行聚类，从而在无需引入后维护的情况下提高查询性能。

连接器使用 Snowflake PIPE 对象作为管理引入的核心组件。PIPE 对象充当所有流数据的入口点和定义层，定义了数据在提交到目标表之前如何进行处理、转换和验证。有关连接器如何与表和管道配合工作的更多信息，请参阅 连接器如何与表和管道配合使用。

选择连接器版本

Kafka Connector 在 Kafka Connect 集群中运行，从 Kafka 主题读取数据并写入 Snowflake 表。

Snowflake 提供两个版本的连接器。两个版本的连接器都为从 Kafka 到 Snowflake 的流式传输数据提供相同的核心功能。

  • 连接器的 Confluent 版本

    Confluent 版本以 zip 文件形式打包，并包含运行连接器所需的所有外部库。如果您使用的是 Confluent Platform，请选择此版本。

    备注

    v4 连接器尚未作为原生 Confluent Cloud 连接器提供。在 Confluent Cloud 上，请将其作为自定义插件连接器进行安装。如需获取 Confluent 安装包，请联系 Snowflake 支持部门。

    有关详细信息，请参阅 Kafka Connect (https://docs.confluent.io/current/connect/)。

  • OSS 连接器的 Apache Kafka 版本

    可从 开源软件 (OSS) Apache Kafka 包 (https://mvnrepository.com/artifact/com.snowflake/snowflake-kafka-connector/) 获取。

    Apache 版本以标准的 fat JAR 文件形式分发，需要手动安装到您的 Apache Kafka Connect 集群中。此版本需要 Bouncy Castle (https://www.bouncycastle.org/) 加密库，必须单独下载。

    有关详细信息，请参阅 Apache Kafka (https://kafka.apache.org/)。

将连接器与 Apache Iceberg™ 表结合使用

连接器可以将数据引入到 Snowflake 管理的 Apache Iceberg™ 表 中。在为 Iceberg 表引入配置 Kafka Connector 之前，必须创建一个 Iceberg 表。有关更多信息，请参阅 创建用于引入的 Apache Iceberg™ 表。

限制

Snowflake Connector for Kafka 有以下限制。

Apache Iceberg™ 表和架构演化

该连接器不支持 Apache Iceberg™ 表的架构演化。

从版本 3.x 迁移现有管道

v4 连接器需要全新的配置（新的连接器类、已移除的属性以及更改后的默认值）。如果配置正确，支持从 Snowpipe 模式和 Snowpipe Streaming 模式进行迁移，且不会出现间隙或重复。切换必须在 ``offsets.retention.minutes``（默认 7 天）内完成。有关详细信息，请参阅 从 v3 迁移到 v4。

单消息转换 (SMTs)：

使用社区转换器时，大多数单消息转换 (SMTs) 均受支持，但 regex.router 目前尚未支持。

身份验证

该连接器仅支持密钥对身份验证。v4 不支持 OAuth。

错误处理行为取决于验证模式

在服务器端验证模式下，损坏的记录会被捕获到 错误表。在客户端验证模式下，连接器在遇到无效记录时会立即停止运行，或者如果已配置 errors.tolerance=all，则会将记录路由到死信队列 (DLQ)。有关详细信息，请参阅 验证和错误处理。

容错限制

可以为 Kafka 主题配置存储空间限制或保留时间限制。

  • 如果系统离线时间超过保留时间，则不会加载过期的记录。同样，如果已超出 Kafka 的存储空间限制，某些消息将无法送达。

  • 如果已删除 Kafka 主题中的消息，这些更改不会反映在 Snowflake 表中。

有关 SMTs 的更多信息，请参阅 针对 Confluent Cloud 或 Confluent Platform 的 Kafka Connect 单消息转换参考 (https://docs.confluent.io/current/connect/transforms/index.html)。

支持的连接器版本

下表描述了支持的连接器版本。

版本系列

状态

备注

4.x.x

正式发布

最新版本。从 3.x 和 2.x 的迁移必须手动完成。

3.x.x

官方支持

建议升级到 v4。

2.xx

官方支持

建议升级。

1.x.x

不支持

备注

正在寻找经典版 Kafka Connector（v3 及更早版本）？请参阅 Kafka Connector v3（经典版）。有关迁移指南，请参阅 从 v3 迁移到 v4。

后续步骤

请审阅 连接器的工作原理 主题，了解有关连接器如何配合表和管道工作的更多信息。 . 请审阅 为 Snowflake Connector for Kafka 设置任务 主题，了解设置 Snowflake Connector for Kafka 的步骤。