设置 Openflow Connector for Kafka¶

先决条件¶

1. 确保您已查看 Openflow Connector for Kafka。

2. 确保您已 设置 Openflow。

连接器类型¶

Openflow Connector for Kafka 有三种不同的配置可供选择，每种配置都针对特定用例进行了优化。您可以从连接器库下载这些连接器定义：

适用于 JSON 数据格式的 Apache Kafka

适用于 JSON 消息引入的简化连接器，具有架构演化和主题到表的映射功能

适用于 AVRO 数据格式的 Apache Kafka

适用于 AVRO 消息引入的简化连接器，具有架构演化和主题到表的映射功能

Apache Kafka with DLQ and metadata

功能齐全的连接器，支持死信队列 (DLQ)、元数据处理，功能与 Snowflake Connector for Kafka 相同

有关特定连接器类型的详细配置，请参阅：

• 适用于 JSON/AVRO 数据格式的 Apache Kafka – JSON/AVRO 数据格式连接器

• 带有 DLQ 和元数据的 Apache Kafka – DLQ 和元数据连接器

设置 Snowflake 账户¶

作为 Snowflake 账户管理员，请执行以下任务：

1. 创建类型为 SERVICE 的新 Snowflake 服务用户。

2. 创建新角色或使用现有角色并授予 数据库权限。

   由于连接器能够自动创建目标表（如果目标表尚不存在），请确保用户具有创建和管理 Snowflake 对象所需的权限：

   对象	权限	备注
   数据库	USAGE
   架构	USAGE . CREATE TABLE .	创建架构级对象后，可以撤销 CREATE object 权限。
   表	OWNERSHIP	仅在使用 Kafka Connector 将数据引入到 现有 表时需要。. 如果连接器为来自 Kafka 主题的记录创建新的目标表，则配置中指定的用户的默认角色将成为表所有者。

   Snowflake 建议为每个 Kafka 实例创建单独的用户和角色，以便更好地进行访问控制。

   您可以使用以下脚本创建和配置自定义角色（需要 SECURITYADMIN 或等效角色）：

   USE ROLE securityadmin;
   
   CREATE ROLE kafka_connector_role_1;
   GRANT USAGE ON DATABASE kafka_db TO ROLE kafka_connector_role_1;
   GRANT USAGE ON SCHEMA kafka_schema TO ROLE kafka_connector_role_1;
   GRANT CREATE TABLE ON SCHEMA kafka_schema TO ROLE kafka_connector_role_1;
   
   -- Only for existing tables
   GRANT OWNERSHIP ON TABLE existing_table1 TO ROLE kafka_connector_role_1;
   

   Copy

   请注意，权限必须直接授予连接器角色且不能继承。

3. 向该 Snowflake 服务用户授予您在前面步骤中创建的角色。

   角色应分配为用户的默认角色：

   GRANT ROLE kafka_connector_role_1 TO USER kafka_connector_user_1;
   ALTER USER kafka_connector_user_1 SET DEFAULT_ROLE = kafka_connector_role_1;
   

   Copy

4. 为第 1 步中创建的 Snowflake SERVICE 用户配置 密钥对身份验证。

5. Snowflake 强烈建议执行此步骤。配置 Openflow 支持的密钥管理器（例如 AWS、Azure 和 Hashicorp），并将公钥和私钥存储在密钥存储库中。

   备注

   如果您出于任何原因不希望使用密钥管理器，则您有责任根据组织的安全策略保护用于密钥对身份验证的公钥和私钥文件。

   1. 配置密钥管理器后，确定如何对其进行身份验证。在 AWS 中，建议您使用与 Openflow 关联的 EC2 实例角色，因为这样就无需保留其他密钥。

   2. 在 Openflow 中，从右上角的汉堡菜单中配置与此密钥管理器关联的参数提供商。导航到 Controller Settings » Parameter Provider，然后提取您的参数值。

   3. 此时，可以使用关联的参数路径引用所有凭据，无需在 Openflow 中保留敏感值。

6. 如果任何其他 Snowflake 用户需要访问连接器引入的原始文档和表（例如，在 Snowflake 中进行自定义处理），则授予这些用户在步骤 1 中创建的角色。

设置连接器¶

作为数据工程师，执行以下任务以安装和配置连接器：

配置特定于变体的设置¶

配置常用参数后，需要配置特定于所选连接器变体的设置：

对于 适用于 JSON 数据格式的 Apache Kafka 和 适用于 AVRO 数据格式的 Apache Kafka 连接器：

请参阅 适用于 JSON/AVRO 数据格式的 Apache Kafka，了解特定于 JSON/AVRO 的参数。

对于 Apache Kafka with DLQ and metadata 连接器：

请参阅 带有 DLQ 和元数据的 Apache Kafka，了解高级参数，其中包括 DLQ 配置、架构化设置、Iceberg 表支持和消息格式选项。

身份验证¶

所有连接器变体都支持通过参数上下文配置的 SASL 身份验证，如 Kafka 源参数（SASL 身份验证） 中所述。

有关其他身份验证方法（包括 mTLS 和 AWS MSK IAM），请参阅 为 Openflow Connector for Kafka 配置其他身份验证方法。

运行流¶

1. 右键点击平面，然后点击 Enable all Controller Services。

2. 右键点击平面，然后点击 Start。连接器开始数据引入。