Snowpipe StreamingでのKafka用Snowflakeコネクタの使用¶

必須のプロパティ¶

Kafkaコネクタのプロパティファイルで接続設定を追加または編集します。詳細については、 Kafkaコネクタの構成 をご参照ください。

snowflake.ingestion.method

ストリーミングインジェストクライアントとしてKafkaコネクタを使用する場合にのみ必要です。 Kafkaトピックデータをロードするために、Snowpipe Streamingまたは標準Snowpipeのいずれを使用するかを指定します。サポートされている値は次のとおりです。

• SNOWPIPE_STREAMING

• SNOWPIPE （デフォルト）

トピックデータをキューに入れてロードするために、バックエンドサービスを選択する追加の設定は必要ありません。通常どおり、Kafkaコネクタのプロパティファイルで追加のプロパティを構成します。

snowflake.role.name

テーブルに行を挿入するときに使用するアクセス制御ロール。

クライアント最適化プロパティ¶

enable.streaming.client.optimization

ワンクライアント最適化を有効にするかどうかを指定します。このプロパティは、Kafkaコネクタのリリースバージョン2.1.2以降でサポートされています。デフォルトで有効になっています。

ワンクライアント最適化では、Kafkaコネクタごとに複数のトピックパーティションに対して1つのクライアントのみが作成されます。この機能により、より大きなファイルを作成することによって、クライアントの実行時間を短縮し、移行コストを削減することができます。

値:

• true

• false

デフォルト:

true

高スループットのシナリオ (例: コネクタあたり50 MB/s) でこのプロパティを有効にすると、遅延やコストの上昇につながる可能性があることに注意してください。スループットの高いシナリオでは、このプロパティを無効にすることをお勧めします。

バッファーおよびポーリングのプロパティ¶

buffer.flush.time

バッファーフラッシュ間の秒数。フラッシュごとに、バッファーされた記録の挿入操作が実行されます。Kafkaコネクタは、各フラッシュの後にSnowpipe Streaming API を1回呼び出します。

buffer.flush.time プロパティでサポートされる最小値は 1 （秒単位）です。データの平均フローレートを高くするには、遅延を低減するためにデフォルト値を下げることをお勧めします。遅延よりもコストが重要な場合は、バッファーのフラッシュ時間を増やすことができます。メモリ不足の例外を回避するため、バッファーがいっぱいになる前にKafkaメモリバッファーをフラッシュするように注意してください。

値:

• 最小: 1

• 最大: 上限なし

デフォルト:

10

Snowpipe Streamingは自動的に1秒ごとにデータをフラッシュしますが、これはKafkaコネクタのバッファフラッシュ時間とは異なることに注意してください。Kafkaバッファのフラッシュ時間に達した後、データはSnowpipe Streamingを通じてSnowflakeに1秒の遅延で送信されます。詳細については、 Snowpipe Streamingの遅延 をご参照ください。

buffer.count.records

Snowflakeにインジェストされる前に、Kafkaパーティションごとにメモリにバッファされる記録の数。

値:

• 最小: 1

• 最大: 上限なし

デフォルト:

10000

buffer.size.bytes

データファイルとしてSnowflakeに取り込まれる前に、Kafkaパーティションごとにメモリーにバッファーされたレコードの累積サイズ（バイト単位）。

レコードは、データファイルに書き込まれるときに圧縮されます。その結果、バッファー内のレコードのサイズは、レコードから作成されたデータファイルのサイズよりも大きくなる可能性があります。

値:

• 最小: 1

• 最大: 上限なし

デフォルト:

20000000 （20 MB）

snowflake.streaming.max.client.lag

Snowflake Ingest Java がSnowflakeにデータをフラッシュする頻度を秒単位で指定します。

値を低くするとレイテンシーは低く保たれますが、特に snowflake.streaming.enable.single.buffer が有効な場合、クエリのパフォーマンスが悪化する可能性があります。詳細情報については、 Snowpipe Streamingの推奨レイテンシー構成 を参照してください。

値:

• 最小: 1 秒

• 最大: 600 秒

デフォルト:

バージョン3.0.0以降では 120 秒、それ以外は 1 秒

snowflake.streaming.enable.single.buffer

Snowpipe Streamingのシングルバッファを有効にして、コネクタの内部バッファでのデータのバッファリングをスキップするかどうかを指定します。

このプロパティは、Kafkaコネクタのバージョン2.3.1以降でサポートされています。

ストリーミングコネクターは、 Snowflake Ingest Java によって提供されるものと一緒に内部バッファを使用します。このプロパティを true に設定すると、Kafkaコネクタは低レイテンシを実現するために内部バッファをスキップします。

このプロパティを true に設定すると、 buffer.flush.time と buffer.count.records は無関係になることに注意してください。

値:

• true

• false

デフォルト:

バージョン3.0.0以降の場合は true、それ以外の場合は false

Kafkaコネクタのプロパティに加えて、Kafkaコンシューマー max.poll.records のプロパティに注意してください。これは、1回のポーリングでKafkaからKafka Connectに返される記録の最大数を制御します。 500 のデフォルト値は増やすことができますが、メモリの制約に注意してください。このプロパティの詳細については、Kafkaパッケージのドキュメントをご参照ください。

• Apache Kafka

• Confluent

エラー処理および DLQ プロパティ¶

errors.tolerance

Kafkaコネクタで発生したエラーの処理方法を指定します。

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

値:

• NONE: 最初のエラーが発生したときにデータのロードを停止します。

• ALL: すべてのエラーを無視して、データのロードを続行します。

デフォルト:

NONE

errors.log.enable

Kafka Connectログファイルにエラーメッセージを書き込むかどうかを指定します。

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

値:

• TRUE: エラーメッセージを書き込みます。

• FALSE: エラーメッセージを書き込みません。

デフォルト:

FALSE

errors.deadletterqueue.topic.name

SnowflakeテーブルにインジェストできなかったメッセージをKafkaに配信するために、Kafkaの DLQ (配信不能キュー) トピック名を指定します。詳細については、 配信不能キュー （このトピック内）をご参照ください。

値:

カスタムテキスト文字列

デフォルト:

なし