Snowflake High Performance connector for Kafka 的工作原理¶

主题名称、表名称和管道名称¶

根据配置设置，连接器将为目标表使用不同的名称。目标表名称始终派生自主题名称。

了解 RECORD_METADATA¶

连接器会用关于 Kafka 记录的元数据填充 RECORD_METADATA 结构。此元数据通过 Snowpipe Streaming 数据源发送至 Snowflake，在管道转换中可通过 $1:RECORD_METADATA 访问器使用。RECORD_METADATA 结构在用户自定义管道和默认管道模式下均可使用。其内容可保存至 VARIANT 类型的列，也可提取其中独立字段并保存到单独的列中。

具有转换和元数据的管道示例：

CREATE PIPE ORDERS AS
COPY INTO ORDERS_TABLE
FROM (
  SELECT
    $1:order_id::NUMBER,
    $1:customer_name,
    $1:order_total,
    $1:RECORD_METADATA.topic AS source_topic,
    $1:RECORD_METADATA.offset::NUMBER AS kafka_offset,
    $1:RECORD_METADATA.SnowflakeConnectorPushTime::BIGINT AS ingestion_time
  FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

在此示例中：

• 管道从 Kafka 消息中提取特定字段（order_id、customer_name、order_total）

• 它还捕获元数据字段（主题、偏移和引入时间戳）

• 可以根据需要对值进行类型转换和/或转换

元数据字段的填充方式¶

连接器会根据 Kafka 记录属性和连接器配置自动填充元数据字段。您可以使用以下配置参数控制包含哪些元数据字段：

• ``snowflake.metadata.topic``（默认值：true）– 包括主题名称

• ``snowflake.metadata.offset.and.partition``（默认值：true）– 包括偏移和分区

• ``snowflake.metadata.createtime``（默认值：true）– 包括 Kafka 记录时间戳

• ``snowflake.metadata.all``（默认值：true）– 包括所有可用元数据

当为 snowflake.metadata.all=true``（默认）时，则填充所有元数据字段。将单个元数据字段设置为 ``false 可将其从 RECORD_METADATA 结构中排除。

RECORD_METADATA 结构默认包含以下信息：

RECORD_METADATA 列中记录的元数据量可使用可选的 Kafka 配置属性进行配置。

字段名称和值区分大小写。

Kafka 记录在引入前如何转换¶

在将每行数据传递给 Snowpipe Streaming 之前，连接器会将 Kafka Connect 记录值转换为一个 Map<String, Object> 类型的对象，其键必须与目标列名匹配（也可在用户自定义管道中进行转换）。基本类型的字符串、字节数组或数字必须进行封装（例如通过 HoistField SMT 方法），以确保连接器接收到结构化对象。转换器应用以下规则：

• Null 值被视为逻辑删除。当 behavior.on.null.values=IGNORE 时它们将被跳过，否则会作为空 JSON 对象被引入。

• 数字和布尔字段按原样传递。精度超过 38 的小数值会被序列化为字符串，以符合 Snowflake 的 NUMBER 限制。

• byte[] 和 ByteBuffer 有效负载是 Base64 编码的字符串，因此将它们存储在 VARIANT 或 VARCHAR 列。

• 数组仍然是数组，嵌套对象仍然是嵌套映射。当您依赖默认管道以原样存储嵌套数据时，需声明 VARIANT 类型的列。

• 由于 Snowflake 列名必须为文本类型，非字符串键的映射将以 [key, value] 键值对数组的形式输出。

• 当相关元数据标志启用时，记录头和键会被复制到 RECORD_METADATA 中。

如果您需要将完整消息正文保留为单列，请使用 SMTs 将其封装到新的顶层字段中。 请参阅转换模式的 旧版 RECORD_CONTENT 列。

用户定义的管道模式¶

在此模式下，您可以完全控制数据转换和列映射。

何时使用此模式：

• 您需要使用与 JSON 字段名不同的自定义列名

• 您需要应用数据转换（类型转换、掩码、筛选）

• 您希望完全控制数据映射到列的方式

默认管道模式¶

在此模式下，连接器使用名为 {tableName}-STREAMING 的默认管道，并将 Kafka 记录字段按名称（不区分大小写）映射到对应的表列。

何时使用此模式：

• 您的 Kafka 记录键名称与所需的列名称相匹配

• 您不需要自定义数据转换

• You want a simple configuration

使用默认管道模式将 Kafka 记录键映射到表列

使用默认管道模式时，连接器会使用名为 {tableName}-STREAMING 的默认管道，并将内容中的一级键通过不区分大小写的匹配方式直接映射到表列。

使用默认管道模式 – 示例¶

{
  "city": "New York",
  "age": 30,
  "married": true,
  "has cat": true,
  "@&$#* includes special characters": true,
  "skills": ["sitting", "standing", "eating"],
  "family": {"son": "Jack", "daughter": "Anna"}
}

CREATE TABLE PERSON_DATA (
  record_metadata VARIANT,
  city VARCHAR,
  age NUMBER,
  married BOOLEAN,
  "has cat" BOOLEAN,
  "!@&$#* includes special characters" BOOLEAN,
  skills VARIANT,
  family VARIANT
);

使用用户定义的管道模式 – 示例¶

本示例展示如何配置并使用具有自定义数据转换功能的用户自定义管道。

CREATE TABLE ORDERS (
  order_id NUMBER,
  customer_name VARCHAR,
  order_total NUMBER,
  order_date TIMESTAMP_NTZ,
  source_topic VARCHAR
);

CREATE PIPE ORDERS AS
COPY INTO ORDERS
FROM (
  SELECT
    $1:order_id::NUMBER,
    $1:customer_name,
    $1:order_total::NUMBER,
    $1:order_date::TIMESTAMP_NTZ,
    $1:RECORD_METADATA.topic
  FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

snowflake.topic2table.map=kafka-orders-topic:ORDERS

{
  "city": "New York",
  "age": 30,
  "married": true,
  "has cat": true,
  " @&$#* has Łułósżź": true,
  "skills": ["sitting", "standing", "eating"],
  "family": {"son": "Jack", "daughter": "Anna"}
}

CREATE TABLE PERSON_DATA (
  city VARCHAR,
  age NUMBER,
  married BOOLEAN,
  has_cat BOOLEAN,
  weird_field_name BOOLEAN,
  skills VARIANT,
  family VARIANT
);

CREATE PIPE PERSON_DATA AS
COPY INTO PERSON_DATA
FROM (
  SELECT
    $1:city,
    $1:age,
    $1:married,
    $1['has cat'] AS has_cat,
    $1[' @&$#* has Łułósżź'] AS weird_field_name,
    $1:skills,
    $1:family
  FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

显式主题到表映射模式¶

当您配置 snowflake.topic2table.map 参数时，连接器在显式映射模式下运行。此模式允许您执行以下操作：

• 将多个 Kafka 主题映射到单个 Snowflake 表

• 使用与主题名称不同的自定义表名称

• 应用正则表达式模式来匹配多个主题

配置格式：

snowflake.topic2table.map 参数接受以逗号分隔的主题到表的映射列表，格式如下：

topic1:table1,topic2:table2,topic3:table3

配置示例：

直接主题映射

snowflake.topic2table.map=orders:ORDER_TABLE,customers:CUSTOMER_TABLE

正则表达式模式匹配

snowflake.topic2table.map=.*_cat:CAT_TABLE,.*_dog:DOG_TABLE

此配置将所有以 _cat 结尾的主题（例如 orange_cat、calico_cat）映射到 CAT_TABLE 表，并将所有以 _dog 结尾的主题映射到 DOG_TABLE 表。

一个表中的多个主题

snowflake.topic2table.map=topic1:shared_table,topic2:shared_table,topic3:other_table

此配置将 topic1 和 topic2 映射到 shared_table，同时将 topic3 映射到 other_table。

旧版 RECORD_CONTENT 列¶

In prior versions of the connector (3.x and earlier), when the schematization feature was disabled, the connector created a destination table with two columns: RECORD_CONTENT and RECORD_METADATA. The RECORD_CONTENT column contained the entire Kafka message content in a column of type VARIANT. The RECORD_METADATA column continues to be supported but the RECORD_CONTENT column is no longer created by the connector. The same functionality can be achieved using SMT transformations (see examples later in this section). The RECORD_CONTENT key is also no longer available in PIPE transformations. For example, this PIPE definition will not work by default:

CREATE PIPE ORDERS AS
COPY INTO ORDERS
FROM (
  SELECT
    $1:RECORD_CONTENT
FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

如果您需要将完整的 Kafka 消息内容保存到单个列中，或者需要在 PIPE 转换中获取整个 Kafka 消息内容，可以使用以下 SMT 转换，将整个 Kafka 消息内容封装到您所需的自定义字段：

transforms=wrapKafkaMessageContent
transforms.wrapKafkaMessageContent.type=org.apache.kafka.connect.transforms.HoistField$Value
transforms.wrapKafkaMessageContent.field=your_top_level_field_name

此转换会将整个 Kafka 消息内容封装到名为 your_top_level_field_name 的自定义字段中。随后，您可以在 PIPE 转换中使用 $1:your_top_level_field_name 访问器来获取整个 Kafka 消息内容。

CREATE PIPE ORDERS AS
COPY INTO ORDERS
FROM (
  SELECT
    $1:your_top_level_field_name
FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

或者，如果您想通过默认管道将完整的元数据和内容保存到单张表中，则无需创建自定义管道；只需创建一个包含两列的表：RECORD_CONTENT 和 your_top_level_field_name。

CREATE TABLE ORDERS (
  record_metadata VARIANT,
  your_top_level_field_name VARIANT
);

要阅读有关 HoistField$Value 转换的更多信息，请参阅 Kafka 文档 (https://kafka.apache.org/39/documentation.html#connect_transforms)。

处理流通道错误和死信队列¶

The connector inspects the Snowpipe Streaming channel status before committing offsets in Kafka. If the connector detects that the rowsErrorCount property on channel has increased since the connector was started, it raises a fatal error (ERROR_5030) when errors.tolerance=none so that data issues don't go unnoticed. To allow ingestion to continue while triaging bad rows, set errors.tolerance=all

errors.tolerance=all