Set up Openflow Connector for Amazon Kinesis Data Streams¶

Voraussetzungen¶

1. Lesen Sie Openflow Connector for Amazon Kinesis Data Streams.

2. Ensure that you have Openflow einrichten – 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¶

Als AWS-Administrator führen Sie die folgenden Aktionen in Ihrem AWS-Konto durch:

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 in the AWS documentation.

2. Ensure that the AWS user has configured Access Key credentials.

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

   Bemerkung

   • 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-Konto einrichten¶

Als Snowflake-Kontoadministrator führen Sie die folgenden Aufgaben aus:

1. Erstellen Sie einen neuen Benutzer für den Snowflake-Dienst mit dem Typ 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:

   Objekt	Berechtigung	Anmerkungen
   Datenbank	USAGE
   Schema	USAGE
   Tabelle	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.

   Sie können das folgende Skript verwenden, um eine benutzerdefinierte Rolle zu erstellen und zu konfigurieren (erfordert SECURITYADMIN oder gleichwertig):

   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

   Bemerkung

   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 empfiehlt diesen Schritt dringend. Konfigurieren Sie einen von Openflow unterstützten Geheimnismanager, z. B. AWS, Azure und Hashicorp, und speichern Sie die öffentlichen und privaten Schlüssel im Geheimnisspeicher.

   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. Zu diesem Zeitpunkt können alle Anmeldeinformationen mit den zugehörigen Parameterpfaden referenziert werden, und es müssen keine sensiblen Werte innerhalb von Openflow aufbewahrt werden.

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.

Einrichten des Konnektors¶

Als Data Engineer führen Sie die folgenden Aufgaben aus, um den Konnektor zu installieren und zu konfigurieren:

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.

Betrachten Sie zum Beispiel die folgende Meldung:

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