Openflow Connector for Kinesis の JSON データ形式を設定する¶

前提条件¶

1. Openflow Connector for Kinesis について を確認します。

2. BYOCを使用してOpenflowを設定 しているか、 Snowflakeデプロイメントを使用してOpenflowを設定 していることを確認します。

3. |OFSFSPCS-plural|を使用する場合、必要なドメインの構成 を精査し、 Kinesis コネクタに必要なドメインへのアクセス許可を付与していることを確認してください。

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

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

1. Kinesis Streamsと DynamoDB <https://docs.aws.amazon.com/streams/latest/dev/kcl-iam-permissions.html>`_ にアクセスするための `IAM 権限を持つ AWS ユーザーがいることを確認します。

2. AWS ユーザーが `アクセスキーの認証情報<https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_access-keys.html>`_ を構成したことを確認します。

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

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

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

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

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

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

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

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

      Copy

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

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

   GRANT ROLE kinesis_connector_role TO USER kinesis_connector_user;
   ALTER USER kinesis_connector_user SET DEFAULT_ROLE = kinesis_connector_role;
   

   Copy

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

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

   注釈

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

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

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

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

7. 他のSnowflakeユーザーがインジェストされたデータと作成されたテーブルへのアクセスを必要とする場合（たとえば、Snowflakeのカスタム処理のため）、ステップ2で作成したロールをそれらのユーザーに付与します。

コネクタを設定する¶

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

パラメーター¶

このセクションでは、 Openflow Connector for Kinesis の JSON データ形式のすべてのパラメーターについて説明します。

コネクタはいくつかのモジュールで構成されています。セットを表示するには、コネクタプロセスグループをダブルクリックします。モジュールのパラメーターコンテキストで各モジュールのパラメーターを設定できます。

フローを実行する¶

1. 平面を右クリックして、 Enable all Controller Services を選択します。

2. コネクタのプロセスグループを右クリックし、 Start を選択します。

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

Icebergテーブルのサポート¶

Openflow Connector for Kinesis は、Iceberg有効 が true に設定されている場合、Snowflake管理の Apache Iceberg™ にデータを取り込むことができます。

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    kinesisMetadata OBJECT()
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table'
  ENABLE_SCHEMA_EVOLUTION = true;

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

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,
    body_temperature FLOAT,
    name STRING,
    approved_coffee_types ARRAY(STRING),
    animals_possessed OBJECT(dogs BOOLEAN, cats BOOLEAN),
    date_added DATE
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table';