Snowflake High Performance connector for Kafka: Install and configure¶

Installing the connector for Confluent¶

为开源 Apache Kafka 安装连接器¶

本节说明了如何为开源 Apache Kafka 安装和配置 Kafka Connector。

分布式模式¶

创建 Kafka 配置文件，例如：file:<path>/<config_file>.json。使用所有连接器配置信息填充该文件。该文件必须为 JSON 格式。

Sample configuration file

{
  "name":"XYZCompanySensorData",
  "config":{
      "connector.class": "com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector",
      "tasks.max": "1",
      "snowflake.topic2table.map": "topic1:table_1,topic2:table_2",
      "snowflake.url.name": "myorganization-myaccount.snowflakecomputing.cn:443",
      "snowflake.warehouse.name": "WH",
      "snowflake.private.key": "-----BEGIN PRIVATE KEY-----\n .... \n-----END PRIVATE KEY-----\n",
      "snowflake.schema.name": "MY_SCHEMA",
      "snowflake.database.name": "MY_DATABASE",
      "snowflake.role.name": "MY_ROLE",
      "snowflake.user.name": "MY_USER",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "key.converter": "org.apache.kafka.connect.storage.StringConverter",
      "errors.log.enable": "true",
      "topics": "topic1,topic2",
      "value.converter.schemas.enable": "false",
      "errors.tolerance": "none"
      }
}

独立模式¶

创建一个配置文件，例如：file:<kafka_dir>/config/SF_connect.properties。使用所有连接器配置信息填充该文件。

Sample configuration file

connector.class=com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector
tasks.max=1
snowflake.topic2table.map=topic1:table_1,topic2:table_2
snowflake.url.name=myorganization-myaccount.snowflakecomputing.cn:443
snowflake.warehouse.name=WH
snowflake.private.key=-----BEGIN PRIVATE KEY-----\n .... \n-----END PRIVATE KEY-----\n
snowflake.schema.name=MY_SCHEMA
snowflake.database.name=MY_DATABASE
snowflake.role.name=MY_ROLE
snowflake.user.name=MY_USER
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter=org.apache.kafka.connect.storage.StringConverter
errors.log.enable=true
topics=topic1,topic2
name=XYZCompanySensorData
value.converter.schemas.enable=false
errors.tolerance=none

所需属性¶

name

应用程序名称。该名称在客户使用的所有 Kafka Connector 中必须是唯一的。此名称必须是有效的 Snowflake 无引号标识符。有关有效标识符的信息，请参阅 标识符要求。

connector.class

com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector

topics

以逗号分隔的主题列表。默认情况下，Snowflake 会假定表名称与主题名称相同。如果表名称与主题名称不一致，则使用可选的 topic2table.map 参数（如下）指定从主题名称到表名称的映射。表名称必须是有效的 Snowflake 无引号标识符。有关有效表名称的信息，请参阅 标识符要求。

备注

需要使用 topics 或 topics.regex 中的 任一项 ；而不是两者。

topics.regex

这是一个正则表达式（“regex”），用于指定包含要加载到 Snowflake 表中的消息的主题。连接器会从与 regex 匹配的任何主题名称中加载数据。regex 必须遵循 Java 正则表达式的规则（即与 java.util.regex.Pattern 兼容）。配置文件应包含 topics 或 topics.regex 中的 任一项，而不是两者。

snowflake.url.name

URL 用于访问您的 Snowflake 账户。URL 必须包括您的 账户标识符。请注意，协议 (https://) 和端口号是可选的。

snowflake.user.name

Snowflake 账户的用户登录名。

snowflake.role.name

连接器将用于向表中插入数据的角色名称。

snowflake.private.key

用于验证用户身份的私钥。仅包含密钥，不包含标头或页脚。如果密钥分为多行，请删除换行符。您可以提供未加密的密钥，也可以提供加密的密钥并提供 snowflake.private.key.passphrase 参数，允许 Snowflake 解密密钥。当且仅当 snowflake.private.key 参数值已加密时才使用此参数。这将解密根据 密钥对身份验证和密钥对轮换 中的说明进行加密的私钥。

备注

另请参阅 可选属性 中的 snowflake.private.key.passphrase。

snowflake.database.name

包含要插入行的表的数据库名称。

snowflake.schema.name

包含要插入行的表的架构名称。

header.converter

仅当记录采用 Avro 格式且包含标头时才需要。其值为 "org.apache.kafka.connect.storage.StringConverter"。

key.converter

这是 Kafka 记录的密钥转换器（例如 "org.apache.kafka.connect.storage.StringConverter"）。Kafka Connector 不使用这个转换器，但 Kafka Connect 平台需要它。

有关当前限制的信息，请参阅 Kafka Connector 限制。

value.converter

该连接器支持标准的 Kafka 社区转换器。根据您的数据格式选择合适的转换器：

• 对于 JSON 记录："org.apache.kafka.connect.json.JsonConverter"

• 对于具有架构注册表的 Avro 记录："io.confluent.connect.avro.AvroConverter"

有关当前限制的信息，请参阅 Kafka Connector 限制。

可选属性¶

snowflake.private.key.passphrase

如果此参数的值不为空，那么该连接器会使用此短语来尝试解密私钥。

tasks.max

任务数，通常与Kafka Connect集群中工作节点的 CPU 内核数相同。为了实现最佳性能，Snowflake 建议将任务数量设置为等于 Kafka 分区的总数，但不超过 CPU 个核心的数量。任务数量高可能会导致内存使用增加和频繁的重新平衡。

snowflake.topic2table.map

此可选参数允许用户指定哪些主题应映射到哪些表。每个主题及其表名称要用冒号隔开（请参阅下面的示例）。表名称必须是有效的 Snowflake 无引号标识符。有关有效表名称的信息，请参阅 标识符要求。主题配置允许使用正则表达式定义主题，就像使用 topics.regex 一样。正则表达式不能含糊不清，任何匹配的主题只能匹配单个目标表。

示例：

topics="topic1,topic2,topic5,topic6"
snowflake.topic2table.map="topic1:low_range,topic2:low_range,topic5:high_range,topic6:high_range"

Copy

可以写成：

topics.regex="topic[0-9]"
snowflake.topic2table.map="topic[0-4]:low_range,topic[5-9]:high_range"

Copy

value.converter.schema.registry.url

如果采用 Avro 格式并且您正在使用架构注册表服务，则此值应为架构注册服务的 URL。否则该字段应为空。

value.converter.break.on.schema.registry.error

如果从架构注册表服务加载 Avro 数据，此属性确定 Kafka Connector 在获取架构 ID 时遇到错误时是否应停止使用记录。默认值为 false。将值设为 true 可启用此行为。

jvm.proxy.host

要使 Snowflake Kafka Connector 能够通过代理服务器访问 Snowflake，请设置此参数以指定代理服务器的主机。

jvm.proxy.port

要使 Snowflake Kafka Connector 能够通过代理服务器访问 Snowflake，请设置此参数以指定代理服务器的端口。

snowflake.streaming.max.client.lag

Specifies how often the connector flushes the data to Snowflake, in seconds.

值:

• 最小值：1 秒

• 最大值：600 秒

默认值:

1 秒

jvm.proxy.username

通过代理服务器进行身份验证的用户名。

jvm.proxy.password

通过代理服务器进行身份验证的用户名的密码。

snowflake.jdbc.map

示例："snowflake.jdbc.map": "networkTimeout:20,tracing:WARNING"

不会验证附加的 JDBC 属性（请参阅 JDBC 驱动程序连接参数参考）。这些附加属性不会验证，且不得覆盖或替换必需的属性，例如：jvm.proxy.xxx、snowflake.user.name、snowflake.private.key、snowflake.schema.name 等。

指定以下任一组合：

• tracing 属性以及 JDBC_TRACE 环境变量

• database 属性以及 snowflake.database.name

将导致模糊行为，并且行为将由 JDBC 驱动程序确定。

value.converter.basic.auth.credentials.source

如果使用 Avro 数据格式并需要安全访问 Kafka架构注册表，请将此参数设置为字符串“USER_INFO”，并设置如下所述的 value.converter.basic.auth.user.info 参数。否则，省略该参数。

value.converter.basic.auth.user.info

如果使用 Avro 数据格式并需要安全访问 Kafka架构注册表，请将此参数设置为字符串“<user_ID>:<password>”，并设置如上所述的 value.converter.basic.auth.credentials.source 参数。否则，省略该参数。

snowflake.metadata.createtime

如果值设置为 FALSE，从 RECORD_METADATA 列的元数据中将省略 CreateTime 属性值。默认值为 TRUE。

snowflake.metadata.topic

如果值设置为 FALSE，从 RECORD_METADATA 列的元数据中将省略 topic 属性值。默认值为 TRUE。

snowflake.metadata.offset.and.partition

如果值设置为 FALSE，从 RECORD_METADATA 列的元数据中将省略 Offset 和 Partition 属性值。默认值为 TRUE。

snowflake.metadata.all

如果值设置为 FALSE， RECORD_METADATA 列中的元数据将完全为空。默认值为 TRUE。

transforms

指定跳过 Kafka Connector 遇到的逻辑删除记录，不将它们加载到目标表中。逻辑删除记录定义为整个值字段为 null 的记录。

将属性值设置为 "tombstoneHandlerExample"。

备注

此属性仅用于 Kafka 社区转换器（即 value.converter 属性值）（例如 org.apache.kafka.connect.json.JsonConverter 或 org.apache.kafka.connect.json.AvroConverter）。要使用 Snowflake 转换器管理逻辑删除记录处理，请改为使用 behavior.on.null.values 属性。

transforms.tombstoneHandlerExample.type

设置 transforms 属性时需要。

将属性值设置为 "io.confluent.connect.transforms.TombstoneHandler"

behavior.on.null.values

指定 Kafka Connector 应如何处理逻辑删除记录。逻辑删除记录定义为整个值字段为 null 的记录。 对于 Snowpipe，Kafka Connector 版本 1.5.5 及更高版本支持此属性。对于 Snowpipe Streaming，Kafka Connector 版本 2.1.0 及更高版本支持此属性。

此属性支持以下值：

DEFAULT

当 Kafka Connector 遇到逻辑删除记录时，它会在内容列中插入一个空的 JSON 的字符串。

IGNORE

Kafka Connector 会跳过逻辑删除记录，并且不会为这些记录插入行。

默认值为 DEFAULT。

备注

逻辑删除记录的引入因引入方法而异：

• 对于 Snowpipe，Kafka Connector 仅使用 Snowflake 转换器。要使用 Kafka 社区转换器管理逻辑删除记录处理，请改为使用 transform 和 transforms.tombstoneHandlerExample.type 属性。

• 对于 Snowpipe Streaming，Kafka Connector 仅使用社区转换器。

发送到 Kafka Broker 的记录不得为 NULL，因为这些记录会被 Kafka Connector 删除，从而导致偏移丢失。在特定用例中，缺失的偏移会破坏 Kafka Connector。建议您使用逻辑删除记录而不是 NULL 记录。

使用密钥对身份验证和密钥轮换¶

Kafka 连接器依赖于密钥对身份验证，而不是用户名和密码身份验证。RSA 使用通过 加密的公钥/私钥对签名。使用 OpenSSL 生成公钥-私钥对。公钥分配给配置文件中定义的 Snowflake 用户。

在完成此页面上的密钥对身份验证任务以及 密钥对轮换 任务后，评估本主题后面关于 外部化密钥 的建议。

要配置公钥/私钥对，请执行以下操作：

1. 在终端窗口中的命令行生成私钥。

   您可以生成加密版或未加密版的私钥。

   备注

   Kafka Connector 支持经过验证且符合联邦信息处理标准 (140-2)（即 FIPS 140-2）的加密算法要求。有关更多信息，请参阅 FIPS 140-2 (https://csrc.nist.gov/publications/detail/fips/140/2/final)。

   要生成未加密的版本，请使用以下命令：

   $ openssl genrsa -out rsa_key.pem 2048
   

   Copy

   要生成加密的版本，请使用以下命令：

   $ openssl genrsa 2048 | openssl pkcs8 -topk8 -v2 <algorithm> -inform PEM -out rsa_key.p8
   

   Copy

   其中 <algorithm> 是符合 FIPS 140-2 标准的加密算法。

   例如，指定 AES 256 作为加密算法：

   $ openssl genrsa 2048 | openssl pkcs8 -topk8 -v2 aes256 -inform PEM -out rsa_key.p8
   

   Copy

   如果生成加密版的私钥，请记录加密短语。稍后，您将在 Kafka 配置文件 snowflake.private.key.passphrase 属性中指定该加密短语。

   示例 PEM 私钥

   -----BEGIN ENCRYPTED PRIVATE KEY-----
   MIIE6TAbBgkqhkiG9w0BBQMwDgQILYPyCppzOwECAggABIIEyLiGSpeeGSe3xHP1
   wHLjfCYycUPennlX2bd8yX8xOxGSGfvB+99+PmSlex0FmY9ov1J8H1H9Y3lMWXbL
   ...
   -----END ENCRYPTED PRIVATE KEY-----
   

   Copy

2. 从命令行中，通过引用私钥生成公钥：

   假设私钥已加密并包含在名为 rsa_key.p8 的文件中，使用以下命令：

   $ openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub
   

   Copy

   示例 PEM 公钥

   -----BEGIN PUBLIC KEY-----
   MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAy+Fw2qv4Roud3l6tjPH4
   zxybHjmZ5rhtCz9jppCV8UTWvEXxa88IGRIHbJ/PwKW/mR8LXdfI7l/9vCMXX4mk
   ...
   -----END PUBLIC KEY-----
   

   Copy

3. 将公钥和私钥文件复制到本地目录进行存储。记下文件的路径。私钥使用 PKCS#8（公钥加密标准）格式存储，并使用您在上一步中指定的加密短语进行加密；但仍应使用操作系统提供的文件权限机制保护文件，防止未经授权的访问。不使用文件时，用户有责任保护该文件的安全。

4. 登录 Snowflake。使用 ALTER USER 将公钥分配给 Snowflake 用户。

   For example:

   ALTER USER jsmith SET RSA_PUBLIC_KEY='MIIBIjANBgkqh...';
   

   Copy

   备注

   • 只有安全管理员（即具有 SECURITYADMIN 角色的用户）或更高级别的用户才能更改用户。

   • 在 SQL 语句中排除公钥标头和页脚。

   使用 DESCRIBE USER 验证用户的公钥指纹：

   DESC USER jsmith;
   +-------------------------------+-----------------------------------------------------+---------+-------------------------------------------------------------------------------+
   | property                      | value                                               | default | description                                                                   |
   |-------------------------------+-----------------------------------------------------+---------+-------------------------------------------------------------------------------|
   | NAME                          | JSMITH                                              | null    | Name                                                                          |
   ...
   ...
   | RSA_PUBLIC_KEY_FP             | SHA256:nvnONUsfiuycCLMXIEWG4eTp4FjhVUZQUQbNpbSHXiA= | null    | Fingerprint of user's RSA public key.                                         |
   | RSA_PUBLIC_KEY_2_FP           | null                                                | null    | Fingerprint of user's second RSA public key.                                  |
   +-------------------------------+-----------------------------------------------------+---------+-------------------------------------------------------------------------------+
   

   Copy

   备注

   RSA_PUBLIC_KEY_2_FP 属性在 配置密钥对轮换 中作了说明。

5. 将整个私钥复制并粘贴到配置文件中的 snowflake.private.key 字段。保存文件。

分布式模式¶

在终端窗口中，执行以下命令：

curl -X POST -H "Content-Type: application/json" --data @<path>/<config_file>.json http://localhost:8083/connectors

独立模式¶

在终端窗口中，执行以下命令：

<kafka_dir>/bin/connect-standalone.sh <kafka_dir>/<path>/connect-standalone.properties <kafka_dir>/config/SF_connect.properties