Openflow Connector for Amazon Kinesis Data Streams を設定する¶

前提条件¶

1. Openflow Connector for Amazon Kinesis Data Streams を確認します。

2. :doc:`/user-guide/data-integration/openflow/setup-openflow-byoc`または:doc:`Openflow のセットアップ - Snowflake デプロイメント</user-guide/data-integration/openflow/setup-openflow-spcs>`があることを確認してください。

3. Openflow - Snowflakeデプロイを使用している場合は、必要なドメインの構成 を確認して、Kinesisコネクタに必要なドメインへのアクセス権を付与していることを確認してください。

AWS での IAM のロールとポリシーの設定¶

AWS 管理者として、 AWS アカウントで以下のアクションを実行します。

1. KinesisデータストリームにアクセスするためにOpenflowが使用する AWS IAM のユーザーまたはロールを作成します。詳細については、AWS ドキュメントの IAM ユーザーの作成 をご参照ください。

2. AWS ユーザーが アクセスキーの認証情報 を構成していることを確認します。

3. AWS ユーザーに次の IAM 権限を付与します。

   サービス	アクション	リソース（ARNs）	目的
   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}	シャードを検出し、共有スループットポーリングを通じて記録を読み取り、ストリーム ARN を解決します。また、強化されたファンアウトコンシューマーを登録し、登録中にコンシューマーのステータスをポーリングします。
   Amazon Kinesis Data Streams	kinesis:DeregisterStreamConsumer、kinesis:DescribeStreamConsumer、kinesis:SubscribeToShard	arn:aws:kinesis:${REGION}:${ACCOUNT_ID}:stream/${STREAM_NAME}/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	チェックポイント/リーステーブル（シャードリース、ノードハートビート、チェックポイント)と、レガシーチェックポイントテーブルからの1回限りの移行時に使用される仮移行テーブルを作成および管理します。

   See moreShow less

   Expand

   IAM ポリシーの例:

   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

   ポリシーの例を使用する前に、次のプレースホルダーを置き換えます。

   プレースホルダー	説明
   ${REGION}	ご自身の AWS リージョン（例: us-east-1）
   ${ACCOUNT_ID}	ご自身の AWS アカウント ID（例: 123456789012）
   ${STREAM_NAME}	**AWS ストリーム名**コネクタパラメーターの値
   ${APPLICATION_NAME}	**AWS Kinesisアプリケーション名**コネクタパラメーターの値。DynamoDB チェックポイントテーブル名および強化されたファンアウト登録済みコンシューマー名として使用されます。

   See moreShow less

   Expand

   注釈

   • ${APPLICATION_NAME}_migration テーブルはレガシーチェックポイントテーブルから新しいスキーマへの1回限りの移行中にのみ作成された DynamoDB の仮テーブルです。移行が完了すると自動的に削除されます。デプロイによりレガシー KCL ベースのコネクタが使用されたことがない場合は、移行テーブルの ARN をポリシーから省略できます。

   • dynamodb:DeleteTable アクションは移行プロセス中に使用され、移行の完了が確認された後にポリシーから削除できます。

   • kinesis:DeregisterStreamConsumer アクションは、プロセッサーがキャンバスから削除されたときに呼び出されます。IAM プリンシパルにこの権限が付与されていない場合、コンシューマーは AWS コンソールまたは CLI を使用して手動で登録解除される必要があります。

Snowflakeアカウントを設定する¶

Snowflakeアカウント管理者として、以下のタスクを実行します。

1. タイプを SERVICE として、新しいSnowflakeサービスユーザーを作成します。

2. 新しいロールを作成するか、既存のロールを使用して データベース権限 を付与します。

   コネクタでは、ユーザーが宛先テーブルを作成する必要があります。ユーザーにSnowflakeオブジェクトの管理に必要な権限が付与されていることを確認します。

   オブジェクト	権限	注意
   データベース	USAGE
   スキーマ	USAGE
   テーブル	OWNERSHIP	コネクタがデータをテーブルに取り込むために必要です。

   See moreShow less

   Expand

   Snowflakeは、アクセス制御を改善するために、各Kinesisストリームに個別のユーザーとロールを作成することをお勧めします。

   以下のスクリプトを使用して、カスタムロールを作成および構成できます（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

   注釈

   権限はコネクタロールに直接付与する必要があり、継承することはできません。

3. 宛先テーブルの構成

   スキーマの変更にサーバー側のスキーマ進化を使用し、DMLエラーのロギングにエラーテーブル を使用することを強くお勧めします。

   以下の例は、テーブルを作成し、OWNERSHIP 権限を追加する方法を示しています。

   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

   これらのコネクタは、スキーマの自動検出と進化をサポートします。Snowflakeのテーブルの構造は、コネクタによってロードされた新しいデータの構造をサポートするために、自動的に定義され、進化します。記録コンテンツの第1レベルのキーを、名前で一致するテーブル列に自動的にマッピングします（大文字と小文字の区別なし）。

   スキーマ進化を有効にすると、Snowflakeは受信ストリームで検出された新しい列を追加し、NOT NULL 制約をドロップすることで、宛先テーブルを自動的に拡張して、新しいデータパターンに対応することができます。詳細については、 テーブルスキーマの進化 をご参照ください。

   ENABLE_SCHEMA_EVOLUTION が有効になっていない場合は、テーブル定義を拡張してスキーマを手動で作成する必要があります。コネクタは、記録コンテンツの第1レベルのキーを名前でテーブル列と一致させようとします。JSONからのキーがテーブル列と一致しない場合、コネクタはキーを無視します。

4. （オプション）シークレットマネージャーの構成

   Snowflakeではこの手順を強く推奨します。Openflowがサポートするシークレットマネージャ（AWS、Azure、Hashicorpなど）を構成し、公開キーと秘密キーを秘密ストアに格納します。

   1. シークレットマネージャを構成したら、その認証方法を決定します。AWS では、他のシークレットを保持する必要がないため、Openflowに関連付けられた EC2 インスタンスロールを使用することをお勧めします。

   2. Openflowキャンバスで、右上のハンバーガーメニューから、このシークレットマネージャーに関連付けられたパラメータープロバイダーを構成します。Controller Settings » Parameter Provider に移動してから、パラメーター値を取得します。

   3. この時点で、すべての認証情報を関連するパラメーターパスで参照することができるため、機密性の高い値をOpenflow内で永続化する必要はありません。

5. ユーザーへのアクセス権の付与

   コネクタによって取り込まれた未加工のデータへのアクセスを必要とする（たとえば、Snowflakeでのカスタム処理のため）他のSnowflakeユーザーについては、ステップ2で作成したロールを付与する必要があります。

コネクタを設定する¶

データエンジニアとして、以下のタスクを実行してコネクタをインストールおよび構成します。

外部ボリュームの使用許可¶

たとえば、Icebergテーブルが kinesis_external_volume 外部ボリュームを使用し、コネクタがロール openflow_kinesis_connector_role_1 を使用する場合、次のステートメントを実行します：

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

取り込み用のApache Iceberg™テーブルの作成¶

コネクタは自動的にIcebergテーブルを作成せず、スキーマの進化もサポートしません。コネクタを実行する前に、Icebergテーブルを手動で作成する必要があります。

Icebergテーブルを作成する場合、Icebergデータタイプ（VARIANTなど）または 互換性のあるSnowflakeデータタイプ を使用できます。

例えば、次のようなメッセージを考えてみましょう：

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

例のメッセージ用にIcebergテーブルを作成するには、以下のステートメントの1つを使用します:

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;