Snowflake High Performance connector for Kafka: Install and configure¶

Installing the connector for Confluent¶

オープンソースApache Kafka用のコネクタのインストール¶

このセクションでは、オープンソースのApache Kafka用にKafkaコネクタをインストールおよび構成する手順について説明します。

分散モード¶

Kafka設定ファイルを作成します（例: <path>/<config_file>.json）すべてのコネクタ構成情報をファイルに入力します。ファイルは JSON 形式にする必要があります。

Sample configuration file

{
  "name":"XYZCompanySensorData",
  "config":{
      "connector.class": "com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector",
      "tasks.max": "1",
      "snowflake.topic2table.map": "topic1:table_1,topic2:table_2",
      "snowflake.url.name": "myorganization-myaccount.snowflakecomputing.com:443",
      "snowflake.warehouse.name": "WH",
      "snowflake.private.key": "-----BEGIN PRIVATE KEY-----\n .... \n-----END PRIVATE KEY-----\n",
      "snowflake.schema.name": "MY_SCHEMA",
      "snowflake.database.name": "MY_DATABASE",
      "snowflake.role.name": "MY_ROLE",
      "snowflake.user.name": "MY_USER",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "key.converter": "org.apache.kafka.connect.storage.StringConverter",
      "errors.log.enable": "true",
      "topics": "topic1,topic2",
      "value.converter.schemas.enable": "false",
      "errors.tolerance": "none"
      }
}

スタンドアロンモード¶

設定ファイルを作成します。例: :file:`<kafka_dir>/config/SF_connect.properties`すべてのコネクタ構成情報をファイルに入力します。

Sample configuration file

connector.class=com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector
tasks.max=1
snowflake.topic2table.map=topic1:table_1,topic2:table_2
snowflake.url.name=myorganization-myaccount.snowflakecomputing.com:443
snowflake.warehouse.name=WH
snowflake.private.key=-----BEGIN PRIVATE KEY-----\n .... \n-----END PRIVATE KEY-----\n
snowflake.schema.name=MY_SCHEMA
snowflake.database.name=MY_DATABASE
snowflake.role.name=MY_ROLE
snowflake.user.name=MY_USER
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter=org.apache.kafka.connect.storage.StringConverter
errors.log.enable=true
topics=topic1,topic2
name=XYZCompanySensorData
value.converter.schemas.enable=false
errors.tolerance=none

必須のプロパティ¶

name

アプリケーション名。これは、顧客が使用するすべてのKafkaコネクタで一意でなければなりません。この名前は、引用符で囲まれていないSnowflakeの有効な識別子である必要があります。有効な識別子については、 識別子の要件 をご参照ください。

connector.class

com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector

topics

トピックのコンマ区切りリスト。デフォルトでは、Snowflakeはテーブル名がトピック名と同じであると想定します。テーブル名がトピック名と同じでない場合は、オプションの topic2table.map パラメーター（下記）を使用して、トピック名からテーブル名へのマッピングを指定します。このテーブル名は、引用符で囲まれていないSnowflakeの有効な識別子である必要があります。有効なテーブル名については、 識別子の要件 をご参照ください。

注釈

両方ではなく、 topics または topics.regex の いずれか が必要です。

topics.regex

これは、Snowflakeテーブルに読み込むメッセージを含むトピックを指定する正規表現（「regex」）です。コネクタは、正規表現に一致するトピック名からデータをロードします。正規表現は、Java正規表現のルールに従う必要があります（つまり、java.util.regex.Patternと互換性があります）。構成ファイルには、両方ではなく、 topics または topics.regex の いずれか を含める必要があります。

snowflake.url.name

Snowflakeアカウントにアクセスするための URL。この URL には、使用する アカウント識別子 が含まれている必要があります。プロトコル（https://）とポート番号はオプションです。

snowflake.user.name

Snowflakeアカウントのユーザーログイン名。

snowflake.role.name

コネクタがテーブルにデータを挿入するために使用するロールの名前。

snowflake.private.key

ユーザーを認証するための秘密キー。ヘッダーまたはフッターではなく、キーのみを含めます。キーが複数の行に分割されている場合、改行を削除します。暗号化されていないキーを提供するか、暗号化されたキーを提供して snowflake.private.key.passphrase パラメーターを提供し、Snowflakeがキーを復号化できるようにします。:emph:` パラメーター値が暗号化されている 時に限り``snowflake.private.key` 、このパラメーターを使用します。これにより、 キーペア認証とキーペアローテーション の指示に従って暗号化された秘密キーが復号化されます。

注釈

オプションのプロパティ の snowflake.private.key.passphrase もご参照ください。

snowflake.database.name

行を挿入するテーブルを含むデータベースの名前。

snowflake.schema.name

行を挿入するテーブルを含むスキーマの名前。

header.converter

レコードがAvroでフォーマットされ、ヘッダーが含まれている場合にのみ必要です。値は "org.apache.kafka.connect.storage.StringConverter" です。

key.converter

これは、Kafkaの記録のキーコンバーターです（例： "org.apache.kafka.connect.storage.StringConverter"）。これはKafkaコネクタでは使用されませんが、Kafka Connectプラットフォームでは必要です。

現在の制限については Kafkaコネクタの制限 をご参照ください。

value.converter

コネクタは、標準のKafkaコミュニティコンバーターをサポートしています。データ形式に基づいて適切なコンバーターを選択します。

• JSON 記録の場合: "org.apache.kafka.connect.json.JsonConverter"

• スキーマレジストリを使用するAvro記録の場合: "io.confluent.connect.avro.AvroConverter"

現在の制限については Kafkaコネクタの制限 をご参照ください。

オプションのプロパティ¶

snowflake.private.key.passphrase

このパラメーターの値が空でない場合、コネクタはこのフレーズを使用して秘密キーの復号化を試みます。

tasks.max

タスクの数。通常、Kafka Connectクラスタのワーカーノード全体の CPU コアの数と同じです。最適なパフォーマンスを実現するために、タスク数をKafkaパーティションの総数と等しく、CPU コアの数を超えないように設定することをお勧めします。タスク数が多いとメモリ消費量が増え、頻繁にリバランスが発生する可能性があります。

snowflake.topic2table.map

このオプションのパラメータにより、ユーザーはどのトピックをどのテーブルにマッピングするかを指定できます。各トピックとそのテーブル名はコロンで区切る必要があります（以下を参照）。このテーブル名は、引用符で囲まれていないSnowflakeの有効な識別子である必要があります。有効なテーブル名については、 識別子の要件 をご参照ください。トピック設定では、 topics.regex の使用と同様に、正規表現を使ってトピックを定義することができます。正規表現はあいまいであってはならず、一致するトピックは単一のターゲットテーブルのみに一致しなければなりません。

例:

topics="topic1,topic2,topic5,topic6"
snowflake.topic2table.map="topic1:low_range,topic2:low_range,topic5:high_range,topic6:high_range"

Copy

次のように書くことができます:

topics.regex="topic[0-9]"
snowflake.topic2table.map="topic[0-4]:low_range,topic[5-9]:high_range"

Copy

value.converter.schema.registry.url

形式がAvroで、スキーマレジストリサービスを使用している場合、これはスキーマレジストリサービスの URL である必要があります。それ以外の場合、このフィールドは空でなければなりません。

value.converter.break.on.schema.registry.error

Schema Registry ServiceからAvroデータをロードする場合、このプロパティは、スキーマIDのフェッチ中にエラーが発生した場合にKafkaコネクタがレコードの消費を停止するかどうかを決定します。デフォルト値は false です。この動作を有効にするには、値を true に設定します。

jvm.proxy.host

Snowflake Kafka Connectorがプロキシサーバー経由でSnowflakeにアクセスできるようにするには、このパラメーターを設定して、そのプロキシサーバーのホストを指定します。

jvm.proxy.port

Snowflake Kafka Connectorがプロキシサーバーを介してSnowflakeにアクセスできるようにするには、このパラメーターを設定してそのプロキシサーバーのポートを指定します。

snowflake.streaming.max.client.lag

Specifies how often the connector flushes the data to Snowflake, in seconds.

値:

• 最小: 1 秒

• 最大: 600 秒

デフォルト:

1 秒

jvm.proxy.username

プロキシサーバーで認証するユーザー名。

jvm.proxy.password

プロキシサーバーで認証するユーザー名のパスワード。

snowflake.jdbc.map

例: "snowflake.jdbc.map": "networkTimeout:20,tracing:WARNING"

その他の JDBC プロパティ（JDBC ドライバーの接続パラメーター参照 参照）は検証されません。これらの追加プロパティは検証されず、以下のような必須プロパティをオーバーライドしたり、その代わりに使ったりしてはなりません: jvm.proxy.xxx 、 snowflake.user.name 、 snowflake.private.key 、 snowflake.schema.name など。

以下の組み合わせのいずれかを指定します:

• tracing プロパティと JDBC_TRACE 環境変数

• database プロパティと snowflake.database.name

曖昧な動作となり、動作は JDBC ドライバーによって決定されます。

value.converter.basic.auth.credentials.source

Avroデータ形式を使用していて、Kafkaスキーマレジストリへの安全なアクセスが必要な場合、このパラメーターを文字列「USER_INFO」に設定し、以下で説明する value.converter.basic.auth.user.info パラメーターを設定します。それ以外の場合は、このパラメーターを省略します。

value.converter.basic.auth.user.info

Avroデータ形式を使用していて、Kafkaスキーマレジストリへの安全なアクセスが必要な場合は、上記のとおりこのパラメーターを文字列「<user_ID>:<password>」に設定し、value.converter.basic.auth.credentials.sourceパラメーターを設定します。それ以外の場合は、このパラメーターを省略します。

snowflake.metadata.createtime

値が FALSE に設定されている場合、 CreateTime プロパティ値は RECORD_METADATA 列のメタデータから省略されます。デフォルト値は TRUE です。

snowflake.metadata.topic

値が FALSE に設定されている場合、 topic プロパティ値は RECORD_METADATA 列のメタデータから省略されます。デフォルト値は TRUE です。

snowflake.metadata.offset.and.partition

値が FALSE に設定されている場合、 Offset および Partition プロパティ値は RECORD_METADATA 列のメタデータから省略されます。デフォルト値は TRUE です。

snowflake.metadata.all

値が FALSE に設定されている場合、 RECORD_METADATA 列のメタデータは完全に空です。デフォルト値は TRUE です。

transforms

Kafkaコネクタが検出したtombstoneレコードをスキップし、それらをターゲットテーブルにロードしないように指定します。tombstoneレコードは、値フィールド全体がnullであるレコードとして定義されます。

プロパティ値を "tombstoneHandlerExample" に設定します。

注釈

このプロパティは、Kafkaコミュニティコンバーター（つまり、 value.converter プロパティ値）でのみ使用します（例: org.apache.kafka.connect.json.JsonConverter または org.apache.kafka.connect.json.AvroConverter）。Snowflakeコンバーターでtombstoneレコードの処理を管理するには、代わりに behavior.on.null.values プロパティを使用します。

transforms.tombstoneHandlerExample.type

transforms プロパティを設定するときに必要です。

プロパティ値を "io.confluent.connect.transforms.TombstoneHandler" に設定

behavior.on.null.values

Kafkaコネクタがtombstoneレコードを処理する方法を指定します。tombstoneレコードは、値フィールド全体がnullであるレコードとして定義されます。 Snowpipe では、このプロパティは Kafkaコネクタのバージョン1.5.5以降でサポートされています。Snowpipeストリーミング では、このプロパティはKafkaコネクタのバージョン2.1.0以降でサポートされています。

このプロパティは、次の値をサポートします。

DEFAULT

Kafkaコネクタがtombstoneレコードを検出すると、コンテンツ列に空の JSON 文字列を挿入します。

IGNORE

Kafkaコネクタはtombstoneレコードをスキップし、これらのレコードの行を挿入しません。

デフォルト値は DEFAULT です。

注釈

Tombstone記録のインジェスチョンは、インジェスチョンメソッドによって異なります。

• Snowpipeでは、KafkaコネクタはSnowflakeコンバーターのみを使用します。Kafkaコミュニティコンバーターを使用してtombstoneレコードの処理を管理するには、代わりに transform プロパティと transforms.tombstoneHandlerExample.type プロパティを使用します。

• Snowpipe Streamingでは、Kafkaコネクタはコミュニティコンバーターのみを使用します。

Kafkaブローカーに送信された記録は、Kafkaコネクタによって削除され、オフセットが欠落するため、 NULL であってはなりません。オフセットの欠落は、特定のユースケースにおいてKafkaコネクタを破壊します。NULL の記録ではなく、tombstoneの記録を使用することをお勧めします。

キーペア認証とキーローテーションの使用¶

Kafkaコネクタは、ユーザー名とパスワード認証の代わりに、キーペア認証に依存しています。この認証方法には、2048ビット（最小）の RSA キーペアが必要です。OpenSSLを使用して公開キーと秘密キーのペアを生成します。公開キーは、構成ファイルで定義されたSnowflakeユーザーに割り当てられます。

このページのキーペア認証タスクと キーペアのローテーション</user-guide/key-pair-auth>`のタスクを完了した後、このトピックの後半にある :ref:`label-kafkahp_externalize_secrets の推奨事項を評価します。

公開/秘密キーペアを構成するには、

1. ターミナルウィンドウのコマンドラインから、秘密キーを生成します。

   秘密キーの暗号化バージョンまたは非暗号化バージョンを生成できます。

   注釈

   Kafkaコネクタは、連邦情報処理標準（140-2）（つまり、 FIPS 140-2）要件を満たすように検証された暗号化アルゴリズムをサポートしています。詳細については、 FIPS 140-2 をご参照ください。

   非暗号化バージョンを生成するには、次のコマンドを使用します。

   $ openssl genrsa -out rsa_key.pem 2048
   

   Copy

   暗号化バージョンを生成するには、次のコマンドを使用します。

   $ openssl genrsa 2048 | openssl pkcs8 -topk8 -v2 <algorithm> -inform PEM -out rsa_key.p8
   

   Copy

   <アルゴリズム> は、 FIPS 140-2準拠の暗号化アルゴリズムです。

   たとえば、暗号化アルゴリズムとして AES 256を指定するには、

   $ openssl genrsa 2048 | openssl pkcs8 -topk8 -v2 aes256 -inform PEM -out rsa_key.p8
   

   Copy

   秘密キーの暗号化されたバージョンを生成する場合、パスフレーズを記録します。後で、Kafka構成ファイルの snowflake.private.key.passphrase プロパティでパスフレーズを指定します。

   サンプル PEM 秘密キー

   -----BEGIN ENCRYPTED PRIVATE KEY-----
   MIIE6TAbBgkqhkiG9w0BBQMwDgQILYPyCppzOwECAggABIIEyLiGSpeeGSe3xHP1
   wHLjfCYycUPennlX2bd8yX8xOxGSGfvB+99+PmSlex0FmY9ov1J8H1H9Y3lMWXbL
   ...
   -----END ENCRYPTED PRIVATE KEY-----
   

   Copy

2. コマンドラインから、秘密キーを参照して公開キーを生成します。

   秘密キーが暗号化され、「 rsa_key.p8 」という名前のファイルに含まれていると仮定して、次のコマンドを使用します。

   $ openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub
   

   Copy

   サンプル PEM 公開キー

   -----BEGIN PUBLIC KEY-----
   MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAy+Fw2qv4Roud3l6tjPH4
   zxybHjmZ5rhtCz9jppCV8UTWvEXxa88IGRIHbJ/PwKW/mR8LXdfI7l/9vCMXX4mk
   ...
   -----END PUBLIC KEY-----
   

   Copy

3. 公開キーファイルと秘密キーファイルを保存用のローカルディレクトリにコピーします。ファイルへのパスを記録します。秘密キーは PKCS#8（公開キー暗号化標準）形式を使用して格納され、前の手順で指定したパスフレーズを使用して暗号化されることに注意してください。ただし、オペレーティングシステムが提供するファイル権限メカニズムを使用して、ファイルを不正アクセスから保護する必要があります。ファイルが使用されていない場合、ファイルを保護するのはユーザーの責任です。

4. Snowflakeにログインします。 ALTER USER を使用して、Snowflakeユーザーに公開キーを割り当てます。

   For example:

   ALTER USER jsmith SET RSA_PUBLIC_KEY='MIIBIjANBgkqh...';
   

   Copy

   注釈

   • ユーザーを変更できるのは、セキュリティ管理者（つまり、 SECURITYADMIN ロールのユーザー）以上のみです。

   • SQL ステートメントで公開キーのヘッダーとフッターを除外します。

   DESCRIBE USER を使用してユーザーの公開キーの指紋を検証します。

   DESC USER jsmith;
   +-------------------------------+-----------------------------------------------------+---------+-------------------------------------------------------------------------------+
   | property                      | value                                               | default | description                                                                   |
   |-------------------------------+-----------------------------------------------------+---------+-------------------------------------------------------------------------------|
   | NAME                          | JSMITH                                              | null    | Name                                                                          |
   ...
   ...
   | RSA_PUBLIC_KEY_FP             | SHA256:nvnONUsfiuycCLMXIEWG4eTp4FjhVUZQUQbNpbSHXiA= | null    | Fingerprint of user's RSA public key.                                         |
   | RSA_PUBLIC_KEY_2_FP           | null                                                | null    | Fingerprint of user's second RSA public key.                                  |
   +-------------------------------+-----------------------------------------------------+---------+-------------------------------------------------------------------------------+
   

   Copy

   注釈

   RSA_PUBLIC_KEY_2_FP プロパティは、 キーペアローテーションの構成 で説明されています。

5. 秘密キー全体をコピーして、構成ファイルの snowflake.private.key フィールドに貼り付けます。ファイルを保存します。

分散モード¶

ターミナルウィンドウから次のコマンドを実行します。

curl -X POST -H "Content-Type: application/json" --data @<path>/<config_file>.json http://localhost:8083/connectors

スタンドアロンモード¶

ターミナルウィンドウから次のコマンドを実行します。

<kafka_dir>/bin/connect-standalone.sh <kafka_dir>/<path>/connect-standalone.properties <kafka_dir>/config/SF_connect.properties