Set up Openflow Connector for Kinesis Data Streams for JSON data format¶

先决条件¶

1. 确保您已查看 关于 Openflow Connector for Kinesis Data Streams。

2. Ensure that you have 设置 Openflow - BYOC or Set up Openflow - Snowflake Deployments.

3. If using Openflow - Snowflake Deployments, ensure that you've reviewed configuring required domains and have granted access to the required domains for the Kinesis connector.

设置 Kinesis 流¶

作为 AWS 管理员，在您的 AWS 账户中执行以下操作：

1. Ensure that you have an AWS User with IAM permissions to access Kinesis Streams and DynamoDB (https://docs.aws.amazon.com/streams/latest/dev/kcl-iam-permissions.html).

2. Ensure that the AWS User has configured Access Key credentials (https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_access-keys.html).

设置 Snowflake 账户¶

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

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

2. 创建目标数据库和目标架构，以创建用于存储数据的目标表。

   1. 如果您计划使用连接器的功能自动创建目标表（如果目标表尚不存在），请确保用户具有创建和管理 Snowflake 对象所需的权限：

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

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

      USE ROLE SECURITYADMIN;
      
      CREATE ROLE kinesis_connector_role;
      GRANT USAGE ON DATABASE kinesis_db TO ROLE kinesis_connector_role;
      GRANT USAGE ON SCHEMA kinesis_schema TO ROLE kinesis_connector_role;
      GRANT CREATE TABLE ON SCHEMA kinesis_schema TO ROLE kinesis_connector_role;
      
      -- Only for existing tables.
      GRANT OWNERSHIP ON TABLE existing_table TO ROLE kinesis_connector_role;
      

      Copy

3. 创建一个类型为 SERVICE 的新 Snowflake 服务用户。

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

   GRANT ROLE kinesis_connector_role TO USER kinesis_connector_user;
   ALTER USER kinesis_connector_user SET DEFAULT_ROLE = kinesis_connector_role;
   

   Copy

5. 为第 3 步中的 Snowflake SERVICE 用户配置 密钥对认证。

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

   备注

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

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

   2. In Openflow, configure a Parameter Provider associated with this Secrets Manager, from the hamburger menu in the upper right. Navigate to Controller Settings » Parameter Provider and then fetch your parameter values.

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

7. If any other Snowflake users require access to the ingested data and created tables (for example, for custom processing in Snowflake), then grant those users the role created in step 2.

设置连接器¶

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

Parameters¶

This section describes all parameters for the Openflow Connector for Kinesis Data Streams for JSON data format.

The connector consists of a several modules. To see the set, double click on the connector process group. You will be able to set the parameters for each module in the module's parameter context.

运行流¶

1. 右键点击“飞机”图标并选择 Enable all Controller Services。

2. Right-click on the connector's process group and select Start.

连接器开始数据引入。

Iceberg table support¶

Openflow Connector for Kinesis Data Streams can ingest data into a Snowflake-managed Apache Iceberg™ table when Iceberg Enabled is set to true.

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    kinesisMetadata OBJECT()
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table'
  ENABLE_SCHEMA_EVOLUTION = true;

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

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    kinesisMetadata OBJECT(
        stream STRING,
        shardId STRING,
        approximateArrival STRING,
        partitionKey STRING,
        sequenceNumber STRING,
        subSequenceNumber INTEGER,
        shardedSequenceNumber STRING
    ),
    id INT,
    body_temperature FLOAT,
    name STRING,
    approved_coffee_types ARRAY(STRING),
    animals_possessed OBJECT(dogs BOOLEAN, cats BOOLEAN),
    date_added DATE
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table';