使用 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 记录键名称与所需的列名称相匹配

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

• 您需要简单配置

使用默认管道模式将 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 列¶

在连接器旧版本（3.x 及更早版本）中，当模式化功能被禁用时，连接器会创建包含两列的目标表：RECORD_CONTENT 和 RECORD_METADATA。RECORD_CONTENT 列以 VARIANT 类型列的形式完整存储 Kafka 消息内容。RECORD_METADATA 列仍受支持，但连接器不再创建 RECORD_CONTENT 列。同样的功能可以通过 SMT 转换来实现（参见本节后续示例）。RECORD_CONTENT 键在 PIPE 转换中也不再可用。例如，默认情况下，此 PIPE 定义不起作用：

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)。

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

连接器在提交 Kafka 偏移量之前，会检查 Snowpipe Streaming 通道状态。如果连接器检测到自连接器启动以来通道上的 rowsErrorCount 属性已增加，它会在 errors.tolerance=none 时引发致命错误 (ERROR_5030)，以防忽视数据问题。要允许在对异常行进行分类时继续进行引入，可设置 errors.tolerance=all

errors.tolerance=all

授予对外部卷的使用权限¶

要将与 Apache Iceberg™ 表关联的外部卷的 USAGE 权限授予 Kafka 连接器所用的角色，请运行以下语句：

例如，如果 Iceberg 表使用 kafka_external_volume 外部卷，而连接器使用角色 kafka_connector_role，请运行以下语句：

USE ROLE ACCOUNTADMIN;
GRANT USAGE ON EXTERNAL VOLUME kafka_external_volume TO ROLE kafka_connector_role;

创建用于引入的 Apache Iceberg™ 表¶

连接器不会自动创建 Iceberg 表，也不支持架构演化。在运行连接器之前，您必须手动创建 Iceberg 表。

创建 Iceberg 表时，您可以使用 Iceberg 数据类型（包括 VARIANT）或 兼容的 Snowflake 类型。

例如，请考虑以下消息：

{
    "id": 1,
    "name": "Steve",
    "body_temperature": 36.6,
    "approved_coffee_types": ["Espresso", "Doppio", "Ristretto", "Lungo"],
    "animals_possessed":
    {
        "dogs": true,
        "cats": false
    },
    "options":
    {
        "can_walk": true,
        "can_talk": false
    },
    "date_added": "2024-10-15"
}

要为示例消息创建 Iceberg 表，请使用以下任一语句：

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    id number(38,0),
    name varchar,
    body_temperature number(4,2),
    approved_coffee_types array(varchar),
    animals_possessed variant,
    options object(can_walk boolean, can_talk boolean),
    date_added date
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table'
  ICEBERG_VERSION = 3;

Copy

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    id INT,
    name string,
    body_temperature float,
    approved_coffee_types array(string),
    animals_possessed variant,
    date_added date,
    options object(can_walk boolean, can_talk boolean),
    )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table'
  ICEBERG_VERSION = 3;

Copy