Set up Openflow Connector for Amazon Kinesis Data Streams¶

先决条件¶

1. Review Openflow Connector for Amazon Kinesis Data Streams.

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

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

Set up IAM roles and policies in AWS¶

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

1. Create an AWS IAM user or role that Openflow will use to access the Kinesis data stream. For more information, see Creating IAM users (https://docs.aws.amazon.com/IAM/latest/UserGuide/id_users_create.html) in the AWS documentation.

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

3. Grant the AWS user the following IAM permissions:

   Service	Actions	Resources (ARNs)	Purpose
   Amazon Kinesis Data Streams	kinesis:DescribeStream, kinesis:DescribeStreamConsumer, kinesis:GetRecords, kinesis:GetShardIterator, kinesis:ListShards, kinesis:RegisterStreamConsumer	arn:aws:kinesis:${REGION}:${ACCOUNT_ID}:stream/${STREAM_NAME}	Discovers shards, reads records through shared-throughput polling, resolves the stream ARN, registers an Enhanced Fan-Out consumer, and polls consumer status during registration.
   Amazon Kinesis Data Streams	kinesis:DeregisterStreamConsumer, kinesis:DescribeStreamConsumer, kinesis:SubscribeToShard	arn:aws:kinesis:${REGION}:${ACCOUNT_ID}:stream/${STREAM_NAME}/consumer/*	Describes, subscribes to, and deregisters Enhanced Fan-Out consumers by consumer ARN.
   Amazon DynamoDB	dynamodb:CreateTable, dynamodb:DeleteTable, dynamodb:DescribeTable, dynamodb:GetItem, dynamodb:PutItem, dynamodb:Query, dynamodb:Scan, dynamodb:UpdateItem	arn:aws:dynamodb:${REGION}:${ACCOUNT_ID}:table/${APPLICATION_NAME}, arn:aws:dynamodb:${REGION}:${ACCOUNT_ID}:table/${APPLICATION_NAME}_migration	Creates and manages the checkpoint/lease table (shard leases, node heartbeats, checkpoints) and a temporary migration table used during one-time migration from legacy checkpoint tables.

   See moreShow less

   Expand

   Example IAM policy:

   CopyExpand

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "KinesisStreamAccess",
               "Effect": "Allow",
               "Action": [
                   "kinesis:DescribeStream",
                   "kinesis:DescribeStreamConsumer",
                   "kinesis:GetRecords",
                   "kinesis:GetShardIterator",
                   "kinesis:ListShards",
                   "kinesis:RegisterStreamConsumer"
               ],
               "Resource": "arn:aws:kinesis:${REGION}:${ACCOUNT_ID}:stream/${STREAM_NAME}"
           },
           {
               "Sid": "KinesisConsumerAccess",
               "Effect": "Allow",
               "Action": [
                   "kinesis:DeregisterStreamConsumer",
                   "kinesis:DescribeStreamConsumer",
                   "kinesis:SubscribeToShard"
               ],
               "Resource": "arn:aws:kinesis:${REGION}:${ACCOUNT_ID}:stream/${STREAM_NAME}/consumer/*"
           },
           {
               "Sid": "DynamoDBTableAccess",
               "Effect": "Allow",
               "Action": [
                   "dynamodb:CreateTable",
                   "dynamodb:DeleteTable",
                   "dynamodb:DescribeTable",
                   "dynamodb:GetItem",
                   "dynamodb:PutItem",
                   "dynamodb:Query",
                   "dynamodb:Scan",
                   "dynamodb:UpdateItem"
               ],
               "Resource": [
                   "arn:aws:dynamodb:${REGION}:${ACCOUNT_ID}:table/${APPLICATION_NAME}",
                   "arn:aws:dynamodb:${REGION}:${ACCOUNT_ID}:table/${APPLICATION_NAME}_migration"
               ]
           }
       ]
   }
   

   Show lessSee more

   Scroll to top

   Before using the example policy, replace the following placeholders:

   Placeholder	描述
   ${REGION}	Your AWS region (for example, us-east-1)
   ${ACCOUNT_ID}	Your AWS account ID (for example, 123456789012)
   ${STREAM_NAME}	The value of the AWS Kinesis Stream Name connector parameter
   ${APPLICATION_NAME}	The value of the AWS Kinesis Application Name connector parameter. Used as the DynamoDB checkpoint table name and as the Enhanced Fan-Out registered consumer name.

   See moreShow less

   Expand

   备注

   • The ${APPLICATION_NAME}_migration table is a temporary DynamoDB table created only during a one-time migration from legacy checkpoint tables to the new schema. It's deleted automatically when migration completes. If your deployment has never used the legacy KCL-based connector, you can omit the migration table ARN from the policy.

   • The dynamodb:DeleteTable action is used during the migration process and can be removed from the policy after migration is confirmed complete.

   • The kinesis:DeregisterStreamConsumer action is invoked when the processor is removed from the canvas. If the IAM principal doesn't have this permission, the consumer must be deregistered manually through the AWS console or CLI.

设置 Snowflake 账户¶

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

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

2. Create a new role or use an existing role and grant the database privileges.

   The connector requires the user to create the destination table. Make sure the user has the required privileges for managing Snowflake objects:

   对象	权限	备注
   数据库	USAGE
   架构	USAGE
   表	OWNERSHIP	Required for the connector to ingest data into a table.

   See moreShow less

   Expand

   Snowflake recommends creating a separate user and role for each Kinesis stream for better access control.

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

   CopyExpand

   USE ROLE securityadmin;
   
   CREATE ROLE openflow_kinesis_connector_role_1;
   GRANT USAGE ON DATABASE kinesis_db TO ROLE openflow_kinesis_connector_role_1;
   GRANT USAGE ON SCHEMA kinesis_schema TO ROLE openflow_kinesis_connector_role_1;
   

   Show lessSee more

   Scroll to top

   备注

   Privileges must be granted directly to the connector role and can't be inherited.

3. Configure the destination table

   We highly recommend using server-side schema evolution for schema changes and an error table for DML error logging.

   The example below shows how to create a table and add OWNERSHIP permissions.

   CopyExpand

   USE ROLE openflow_kinesis_connector_role_1;
   
   CREATE TABLE kinesis_db.kinesis_schema.<DESTINATION_TABLE_NAME> (
     kinesisMetadata object
   )
   ENABLE_SCHEMA_EVOLUTION = TRUE
   ERROR_LOGGING = TRUE;
   
   USE ROLE securityadmin;
   GRANT OWNERSHIP ON TABLE <DESTINATION_TABLE_NAME> TO ROLE openflow_kinesis_connector_role_1;
   

   Show lessSee more

   Scroll to top

   These connectors provide support for automatic schema detection and evolution. The structure of tables in Snowflake is defined and evolved automatically to support the structure of new data loaded by the connector. It will automatically map the record content's first-level keys to table columns matching by name (case-insensitive).

   With Schema evolution enabled, Snowflake can automatically expand the destination table by adding new columns that are detected in the incoming stream and dropping NOT NULL constraints to accommodate new data patterns. For more information, see Table schema evolution.

   If ENABLE_SCHEMA_EVOLUTION is not enabled, then you have to create the schema manually by extending the table definition. The connector tries to match the record content's first-level keys to the table columns by name. If keys from the JSON do not match the table columns, the connector ignores the keys.

4. (Optional) Configure a secrets manager

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

   1. Once the secrets manager is configured, determine how you will authenticate to it. On AWS, it's recommended that you use the EC2 instance role associated with Openflow as this way no other secrets have to be persisted.

   2. In the Openflow canvas, 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 中保留敏感值。

5. Grant access to users

   Any other Snowflake users who require access to the raw ingested data by the connector (for example, for custom processing in Snowflake), should be granted the role created in step 2.

设置连接器¶

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

Grant usage on an external volume¶

For example, if your Iceberg table uses the kinesis_external_volume external volume and the connector uses the role openflow_kinesis_connector_role_1, run the following statement:

USE ROLE ACCOUNTADMIN;
GRANT USAGE ON EXTERNAL VOLUME kinesis_external_volume TO ROLE openflow_kinesis_connector_role_1;

Create an Apache Iceberg™ table for ingestion¶

The connector does not create Iceberg tables automatically and does not support schema evolution. Before you run the connector, you must create an Iceberg table manually.

When you create an Iceberg table, you can use Iceberg data types (including VARIANT) or compatible Snowflake types.

例如，请考虑以下消息：

{
  "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"
}

To create an Iceberg table for the example message, use one of the following statements:

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,
  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;