Set up Openflow Connector for Amazon Kinesis Data Streams¶

Prerequisites¶

1. Review Openflow Connector for Amazon Kinesis Data Streams.

2. Ensure that you have Set up 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¶

As an AWS administrator, perform the following actions in your AWS account:

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	Description
   ${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

   Note

   • 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.

Set up Snowflake account¶

As a Snowflake account administrator, perform the following tasks:

1. Create a new Snowflake service user with the type as SERVICE.

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:

   Object	Privilege	Notes
   Database	USAGE
   Schema	USAGE
   Table	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.

   You can use the following script to create and configure a custom role (requires SECURITYADMIN or equivalent):

   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

   Note

   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 strongly recommends this step. Configure a secrets manager supported by Openflow, for example, AWS, Azure, and Hashicorp, and store the public and private keys in the secret store.

   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. At this point all credentials can be referenced with the associated parameter paths and no sensitive values need to be persisted within 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.

Set up the connector¶

As a data engineer, perform the following tasks to install and configure the connector:

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.

For example, consider the following message:

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