Snowflake Connector for Kafka:インストールと構成¶

Kafkaコネクタファイルのダウンロード¶

次のいずれかの場所からKafkaコネクタ JAR ファイルをダウンロードします。

Confluentハブ:

https://www.confluent.io/hub/

このパッケージには、キーペア認証に暗号化または非暗号化秘密キーを使用するために必要なすべての依存関係が含まれています。詳細については、このトピックの後半にある キーペア認証とキーローテーションの使用 をご参照ください。

Maven Central Repository:

https://mvnrepository.com/artifact/com.snowflake

このバージョンを使用する場合は、`Bouncy Castle <https://www.bouncycastle.org/>`_暗号化ライブラリ（JARファイル）をダウンロードする必要があります。

• https://mvnrepository.com/artifact/org.bouncycastle/bc-fips/2.1.0

• https://mvnrepository.com/artifact/org.bouncycastle/bcpkix-fips/2.1.8

これらのファイルをKafkaコネクタ JAR ファイルと同じローカルフォルダーにダウンロードします。

コネクタのソースコードはhttps://github.com/snowflakedb/snowflake-kafka-connectorから入手できます。

Kafkaコネクタのインストール¶

他のコネクタをインストールするために提供されている手順を使用して、Kafkaコネクタをインストールします。

https://docs.confluent.io/current/connect/userguide.html

Apache Kafkaのインストール¶

1. Kafka公式ウェブサイト からKafkaパッケージをダウンロードします。

2. ターミナルウィンドウで、パッケージファイルをダウンロードしたディレクトリに移動します。

3. 次のコマンドを実行して、:file:`kafka_<scala_version>-<kafka_version>.tgz`ファイルを解凍します。

   CopyExpand

   tar xzvf kafka_<scala_version>-<kafka_version>.tgz
   

   Show lessSee more

   Scroll to top

JDK のインストール¶

Java Development Kit（JDK）バージョン11以降をインストールして構成します。Snowflakeは、 SEのStandard Edition（JDK）でテストします。Enterprise Edition（EE）は互換性があると予想されますが、テストされていません。

以前に JDK をインストールしている場合 、このセクションをスキップできます。

1. `Oracle JDK ウェブサイト<https://www.oracle.com/technetwork/java/javase/downloads/index.html>`_ から JDK をダウンロードします。

2. JDKをインストールまたは解凍します。

3. ご使用のオペレーティングシステムの手順に従って、環境変数 JAVA_HOME が JDKを含むディレクトリを指すように設定します。

Kafka Connector JAR ファイルをダウンロードします¶

1. Maven Central RepositoryからKafkaコネクタ JAR ファイルをダウンロードします。

   https://mvnrepository.com/artifact/com.snowflake

2. `Bouncy Castle<https://www.bouncycastle.org/>`_ 暗号化ライブラリjarファイルをダウンロードします。

   • https://mvnrepository.com/artifact/org.bouncycastle/bc-fips/2.1.0

   • https://mvnrepository.com/artifact/org.bouncycastle/bcpkix-fips/2.1.8

3. Kafkaデータが Apache Avro 形式でストリーミングされている場合、Avro JAR ファイル（1.11.4）をダウンロードします。

   • https://mvnrepository.com/artifact/org.apache.avro/avro

コネクタのソースコードはhttps://github.com/snowflakedb/snowflake-kafka-connectorから入手できます。

Kafkaコネクタのインストール¶

オープンソースApache Kafka用のコネクタのインストール でダウンロードした JAR ファイルを <kafka_dir>/libs フォルダーにコピーします。

スキーマ化と検証プロパティ¶

これらのプロパティは、コネクタがデータを処理する方法と検証する方法を制御します。新規インストールの場合は、デフォルトが適切に機能します。v3から移行する場合は、使用する値のガイダンスについて:doc:`migrate-v3-to-v4`を確認してください。

snowflake.enable.schematization

受信レコードを個別のテーブル列にスキーマ化するか、レガシーVARIANT列にラップするかを制御します。

true``（デフォルト）の場合は、記録フィールドが名前別に個別のテーブル列にマッピングされます。``false``の場合、コネクタは記録を2つのVARIANT列（``RECORD_CONTENT``および``RECORD_METADATA）に保存します。これはv3の動作と一致します。

デフォルト:

true

snowflake.validation

データ検証とスキーマ進化を実行する場所を制御します。

``server_side``（デフォルト）：検証はCOPYおよびSnowpipeの動作と整合するようにSnowflakeバックエンドによって実行されます。無効な記録は:doc:`エラーテーブル</user-guide/snowpipe-streaming/snowpipe-streaming-high-performance-error-handling>`にキャプチャされます。デフォルトのパイプモードとユーザー定義のパイプモードの両方をサポートします。

client_side:コネクタは、Snowflakeに行を送信する前にデータ型とスキーマの互換性を検証します。無効なレコードについて配信不能キュー（DLQ）をサポートします。デフォルトのパイプモードでのみ動作します。

詳細については、 Validation and error handling をご参照ください。

デフォルト:

server_side

移行および互換性のプロパティ¶

これらのプロパティは、v3から移行する場合に関連します。新規インストールの場合は、これらをスキップして``snowflake.streaming.validate.compatibility.with.classic=false``を設定できます。

snowflake.compatibility.enable.autogenerated.table.name.sanitization

自動生成されたテーブル名がトピック名から派生される方法を制御します。

``false``（デフォルト）の場合、トピック名はテーブル名にそのまま使用され、大文字および小文字の区別と特殊文字が保持されます。テーブル名は引用符で囲まれた識別子として作成されます。

``true``の場合、無効なSnowflake識別子文字はアンダースコアに置き換えられ、名前は大文字で表記されて、一意性を確保するためにハッシュコードが追加されます。これはv3の動作と一致します。

デフォルト:

false

snowflake.compatibility.enable.column.identifier.normalization

列識別子の処理方法を制御します。

``false``（デフォルト）の場合、列識別子は大文字および小文字の区別と特殊文字をそのまま保持します。

``true``の場合、列識別子は大文字に正規化され、v3の動作と一致します。

デフォルト:

false

snowflake.streaming.validate.compatibility.with.classic

すべての移行関連の構成が明示的に設定されているかどうかを確認する起動検証を有効にします。``true``の場合は、以下の構成のいずれかが欠落しているかv3の動作との互換性がない場合に、コネクタは起動時に失敗して状態を説明するエラーが表示されます。

• ``snowflake.validation``は``client_side``であることが必要です

• ``snowflake.compatibility.enable.column.identifier.normalization``は``true``であることが必要です

• ``snowflake.compatibility.enable.autogenerated.table.name.sanitization``は``true``であることが必要です

• ``snowflake.enable.schematization``は明示的に``true``または``false``（デフォルトはv3とv4の間で変更されました）に設定されている必要があります

• ``snowflake.streaming.classic.offset.migration``を明示的に設定する必要があります

• ``snowflake.streaming.classic.offset.migration.include.connector.name``を明示的に設定する必要があります（オフセットの移行が``strict``または``best_effort``の場合）

これにより、互換性を損う変更を確認せずに、コピーされたv3構成でv4を意図せず実行することが防止されます。構成を確認した後にこのチェックをスキップするには、``false``に設定します。

注釈

新規インストール（v3からの移行なし）の場合は、これを``false``に設定します。互換性検証ツールは、既存のv3デプロイからアップグレードする場合にのみ必要です。

詳細については、 Migrate from Kafka connector v3 to v4 をご参照ください。

デフォルト:

true

オフセット移行プロパティ¶

これらのプロパティは、v4がv3 Snowpipe Streaming（SSv1）チャネルからコミットされたオフセットを移行する方法を制御します。``snowflake.ingestion.method=SNOWPIPE_STREAMING``を使用したv3コネクタから移行する場合にのみ関連します。v3 Snowpipeモード（ファイルベースのインジェスチョン）から移行する場合は、``snowflake.streaming.classic.offset.migration``を``skip``に設定します。

snowflake.streaming.classic.offset.migration

v4がv3 Snowpipe Streaming（SSv1）チャネルからオフセットを移行する方法を制御します。

strict: v4は、v3 SSv1チャネルからコミットされたオフセットを検索して、そのポイントから再開します。SSv1チャネルが見つからない場合、コネクタは失敗してエラーが表示されます。

best_effort: v4は、v3 SSv1チャネルからコミットされたオフセットを検索しようとします。チャネルが見つからない場合、v4はKafkaコンシューマーグループのオフセットにフォールバックします。

``skip``（デフォルト）：SSv1オフセットの移行は実行されません。v4は、Kafkaコンシューマーグループのオフセットを使用します。v3 Snowpipeモード（Snowpipe Streamingではありません）から移行する場合は、これを使用します。

デフォルト:

skip

snowflake.streaming.classic.offset.migration.include.connector.name

SSv1チャネル名の検索にコネクタ名が含まれるかどうかを制御します。これは、v3コネクタの構成方法と一致する必要があります。v3では、``snowflake.streaming.channel.name.include.connector.name``プロパティがチャネル名にコネクタ名が含まれるようにするかどうかを制御していました。

v3コネクタに``snowflake.streaming.channel.name.include.connector.name=true``が設定されている場合、またはKafkaコネクタバージョン2.1.0または2.1.1（これらのバージョンにはデフォルトでコネクタ名が含まれていました）を実行している場合は``true``に設定します。それ以外の場合は``false``に設定します。

``snowflake.streaming.classic.offset.migration``が``strict``または``best_effort``である場合にのみ必須です。

デフォルト:

なし（オフセット移行がアクティブな場合は明示的に設定する必要があります）

エラー処理プロパティ¶

errors.tolerance

インジェスチョン中にコネクタがエラーにどのように応答するかを制御します。

``none``（デフォルト）：コネクタタスクは最初のエラーで失敗します。サーバー側の検証では、エラー検出は非同期的であるため、破損した記録の後のいくつかの記録は、タスクが失敗する前に取り込まれる可能性があります。

all:コネクタはデータの取り込みを続行します。クライアント側の検証により、無効な記録はDLQ（構成されている場合）にルーティングされるか、警告が表示されることなくドロップされます。

警告

DLQトピックを構成せずに``errors.tolerance=all``を設定すると、クライアント側の検証を使用する際にトピックの無効な記録が警告が表示されることなくドロップされます。これにより、データが失われる可能性があります。

デフォルト:

none

errors.deadletterqueue.topic.name

配信不能キューのKafkaトピック名。``snowflake.validation=client_side``および``errors.tolerance=all``の場合にのみ有効です。

デフォルト:

空（DLQが無効）

errors.log.enable

``true``の場合、エラーは失敗した操作の詳細と記録プロパティとともにログに記録されます。

デフォルト:

false

enable.task.fail.on.authorization.errors

``true``の場合、Snowflakeからの認証エラーが発生した際にコネクタタスクは直ちに失敗します。``false``の場合、コネクタは再試行します。

デフォルト:

false

プロパティのキャッシュ保存¶

snowflake.cache.table.exists

テーブルの存在チェックのキャッシュ保存を有効にし、Snowflakeへのメタデータクエリの数を削減します。

デフォルト:

true

snowflake.cache.table.exists.expire.ms

テーブル存在チェックに関するキャッシュ有効期限（ミリ秒単位）。

デフォルト:

300000 （5分間）

snowflake.cache.pipe.exists

パイプの存在チェックに関するキャッシュ保存を有効にします。

デフォルト:

true

snowflake.cache.pipe.exists.expire.ms

パイプ存在チェックに関するキャッシュ有効期限（ミリ秒単位）。

デフォルト:

300000 （5分間）

モニタリングと診断のプロパティ¶

jmx

コネクタメトリックのJMX MBeansを有効にします。詳細については、 Snowflake Connector for Kafka をモニターする をご参照ください。

デフォルト:

true

enable.mdc.logging

MDC（Mapped Diagnostic Context）を有効にして、コネクタのコンテキストをログメッセージに追加します。これは、複数のコネクタインスタンスを実行する場合に活用できます。

デフォルト:

false

snowflake.streaming.metadata.connectorPushTime

``true``の場合、``RECORD_METADATA``に``SnowflakeConnectorPushTime``のタイムスタンプが含まれます。このフィールドは、コネクタが取り込みのために記録をバッファしたタイミングを記録し、エンドツーエンドのレイテンシを推定するのに役立ちます。

デフォルト:

true

高度なプロパティ¶

snowflake.streaming.client.provider.override.map

Snowpipe Streamingクライアントプロパティのオーバーライド。形式: key1:value1,key2:value2。Snowflakeサポートに相談した後にのみ使用してください。

デフォルト:

empty

その他のプロパティ¶

snowflake.private.key.passphrase

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

tasks.max

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

snowflake.topic2table.map

``topic:table``形式のトピックからテーブルへのマッピングのコンマ区切りリスト。トピック名の正規表現パターンをサポートします。正規表現はあいまいにはできず、一致するトピックは単一のターゲットテーブルのみに一致しなければなりません。

トピック名とテーブル名は両方とも、二重引用符で囲んで、特殊文字（コロン、コンマ、スペース）をサポートできます。引用符で囲まれていないテーブル名は大文字になります。引用符で囲まれたテーブル名は大文字と小文字の区別を保持します。

正規表現パターン、多対一のマッピング、引用などの詳細な例については、:ref:`明示的なトピックからテーブルへのマッピング<label-kafkahp_explicit_topic_to_table_mapping>`をご参照ください。

例:

CopyExpand

snowflake.topic2table.map=topic1:low_range,topic2:low_range,"my:topic":"My_Table"

Show lessSee more

Scroll to top

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にアクセスできるようにするには、このパラメーターを設定してそのプロキシサーバーのポートを指定します。

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の記録を使用することをお勧めします。

シークレットの外部化¶

Snowflakeは、秘密キーなどの秘密を外部化し、暗号化された形式、または AWS Key Management Service（KMS）、Microsoft Azure Key Vault、 HashiCorp Vaultなどのキー管理サービスで保存することを強くお勧めします。これは、Kafka Connectクラスターで ConfigProvider 実装を使用して実現できます。

詳細については、この サービス のConfluent説明をご参照ください。