Openflow Connector for Kafkaを設定する¶

前提条件¶

1. Openflow Connector for Kafka を確認してください。

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

コネクタタイプ¶

Openflow Connector for Kafkaは3つの異なる構成で利用可能です。それぞれ特定のユースケースに最適化されています。これらのコネクタ定義は、コネクタギャラリーからダウンロードできます。

Apache Kafka for JSON データ形式

スキーマの進化とトピックからテーブルへのマッピングを備えた JSON メッセージ取り込み用の簡易コネクタ

Apache Kafka for AVRO データ形式

スキーマの進化とトピックからテーブルへのマッピングを備えた AVRO メッセージ取り込み用の簡易コネクタ

DLQ とメタデータを使用したApache Kafka

配信不能キュー（DLQ）サポート、メタデータ処理、機能パリティ :doc:`Kafka用Snowflakeコネクタ </user-guide/kafka-connector-overview>`のあるフル機能のコネクタ

特定のコネクタタイプの詳しい構成については、次をご参照ください。

• JSON/AVRO データ形式用Apache Kafka - JSON/AVRO データ形式コネクタ

• DLQ とメタデータを使用したApache Kafka - DLQ およびメタデータコネクタ

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

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

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

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

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

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

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

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

   USE ROLE securityadmin;
   
   CREATE ROLE kafka_connector_role_1;
   GRANT USAGE ON DATABASE kafka_db TO ROLE kafka_connector_role_1;
   GRANT USAGE ON SCHEMA kafka_schema TO ROLE kafka_connector_role_1;
   GRANT CREATE TABLE ON SCHEMA kafka_schema TO ROLE kafka_connector_role_1;
   
   -- Only for existing tables
   GRANT OWNERSHIP ON TABLE existing_table1 TO ROLE kafka_connector_role_1;
   

   Copy

   権限はコネクタロールに直接付与する必要があり、継承できないことに注意してください。

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

   ロールは、ユーザーの既定のロールとして割り当てる必要があります：

   GRANT ROLE kafka_connector_role_1 TO USER kafka_connector_user_1;
   ALTER USER kafka_connector_user_1 SET DEFAULT_ROLE = kafka_connector_role_1;
   

   Copy

4. ステップ1の Snowflake SERVICEユーザーに対して、キーペア認証 を構成します。

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

   注釈

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

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

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

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

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

コネクタを設定する¶

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

バリアント固有の設定を構成する¶

共通パラメーターを構成した後、選択したコネクタバリアントに固有の設定を構成する必要があります。

Apache Kafka for JSON データ形式 および Apache Kafka for AVRO データ形式 コネクタの場合:

JSON/AVRO 固有のパラメーターについては、 JSON/AVRO データ形式用Apache Kafka をご参照ください。

DLQ とメタデータ コネクタを持つ Apache Kafkaの場合:

DLQ 構成、スキーマ化の設定、Icebergテーブルのサポート、メッセージ形式オプションなどの高度なパラメーターについては、 DLQ とメタデータを使用したApache Kafka をご参照ください。

認証¶

Kafkaソースパラメーター（SASL 認証） で説明されているように、すべてのコネクタバリアントは、パラメーターコンテキストを介して構成された SASL 認証をサポートしています。

mTLS や AWS MSK IAM を含むその他の認証方法については、 Openflow Connector for Kafkaのその他の認証方法を構成する をご参照ください。

フローを実行する¶

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

2. プレーンを右クリックし、 Start をクリックします。コネクタがデータの取り込みを開始します。