Openflow Connector for Kinesis を設定する¶

前提条件¶

1. Openflow Connector for Kinesis について を確認してください。

2. Openflowを設定した ことを確認します。

Kinesisストリームを設定する¶

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

1. IAM 権限を持つ AWS アカウントがあり、Kinesis Streamsおよび DynamoDB にアクセスできることを確認します。

2. オプションで、デッドレターキュー（DLQ）Kinesisストリームを作成します。正常に解析できなかったメッセージは、指定した DLQ にリダイレクトすることができます。

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

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

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

2. データを格納する宛先テーブルを作成するために使用する宛先データベースと宛先スキーマを作成します。

   1. 宛先テーブルがまだ存在しない場合に自動的に作成するコネクタの機能を使用する予定の場合は、ユーザーがSnowflakeオブジェクトの作成および管理に必要な権限を持っていることを確認してください。

      オブジェクト	権限	注意
      データベース	USAGE
      スキーマ	USAGE . CREATE TABLE .	スキーマレベルのオブジェクトが作成された後に、 CREATE object 権限を取り消すことができます。
      テーブル	OWNERSHIP	Kinesisコネクタを使用して 既存の テーブルにデータを取り込む場合にのみ必要です。 . コネクタがKinesisストリームからの記録用に新しいターゲットテーブルを作成する場合、構成で指定されたユーザーの既定のロールがテーブル所有者になります。

      以下のスクリプトを使用して、カスタムロールを作成および構成できます（SECURITYADMIN または同等のロールが必要です）。

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

      Copy

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

4. Snowflakeサービスユーザーに、前の手順で作成したロールを付与します。

   GRANT ROLE kinesis_connector_role_1 TO USER kinesis_connector_user_1;
   ALTER USER kinesis_connector_user_1 SET DEFAULT_ROLE = kinesis_connector_role_1;
   

   Copy

5. ステップ3のSnowflake SERVICE ユーザー向けの キーペア認証 で構成します。

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

   注釈

   何らかの理由でシークレットマネージャを使用したくない場合は、組織のセキュリティポリシーに従って、キーペア認証に使用する公開キーと秘密キーファイルを保護する責任があります。

   1. シークレットマネージャを構成したら、その認証方法を決定します。AWS 上では、Openflowに関連付けられた EC2 インスタンスロールが推奨されます。こうすることで、他の秘密を永続化する必要がなくなるからです。

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

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

7. 他のSnowflakeユーザーが、コネクタによって取り込まれた生の取り込みドキュメントやとテーブルへのアクセスを必要とする場合は（Snowflakeでのカスタム処理のためなど）、それらのユーザーにステップ1で作成したロールを付与します。

8. コネクタが使用するウェアハウスを指定します。まずは最小のウェアハウスサイズから始め、複製するテーブルの数や転送するデータ量に応じて異なるサイズを試してみてください。テーブル数が大きい場合は、通常、ウェアハウスのサイズを大きくするよりも、 マルチクラスターウェアハウス を使用した方がスケーリングが向上します。

コネクタを設定する¶

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

フローを実行する¶

1. プレーンを右クリックし、 Enable all Controller Services を選択します。

2. インポートしたプロセスグループを右クリックし、 Start を選択します。

コネクタがデータの取り込みを開始します。

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

既知の問題¶

• コネクタのプロセスグループには、「Upload Failure」という名前の出力ポートが1つあります。Snowflakeに正常にアップロードされなかった FlowFiles の処理に使用することができます。このポートがコネクタのプロセスグループの外部に接続されていないと警告サインが表示されますが、無視することができます。

• すべてのプロセッサーは、停止しているときに、一度だけ実行するように命令することができます。ConsumeKinesisStream プロセッサーは、内部アーキテクチャのため、一度だけ実行を命じられても有意義な仕事はしません。プロセッサーが動き出すには、始動して約2分間作動させる必要があります。