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

必要最小バージョン¶

Snowpipe StreamingをサポートするKafkaコネクタの最小必要バージョンは2.0.0です。

Kafka構成プロパティ¶

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

必ず1回のセマンティクス¶

必ず1回のセマンティクスにより、重複やデータ損失なしでKafkaメッセージの配信が保証されます。この配信保証は、Snowpipe Streamingを使用するKafkaコネクタに対してデフォルトで設定されています。

Kafkaコネクタは、パーティションとチャネル間の1対1マッピングを採用し、2つの異なるオフセットを使用します。

• コンシューマーオフセット: これは、コンシューマーによって消費された最新のオフセットを追跡し、Kafkaによって管理されます。

• オフセットトークン: これは、Snowflakeでコミットされた最新のオフセットを追跡し、Snowflakeによって管理されます。

Kafkaコネクタは、常にオフセットの欠落を処理するわけではないことに注意してください。Snowflakeは、すべての記録が順次増加するオフセットを持つことを期待しています。オフセットの欠落は、特定のユースケースにおいてKafkaコネクタを破壊します。 NULL の記録ではなく、tombstoneの記録を使用することをお勧めします。

Kafkaコネクタは、次のベストプラクティスを実装することで、1回限りの配信を実現します。

チャネルの開始/再開:

• 特定のパーティションのチャネルを開始したり、再開したりする場合、Kafkaコネクタは、 getLatestCommittedOffsetToken API を通じてSnowflakeから取得した、最新のコミットされたオフセットトークンを信頼できる情報源として使用し、それに応じてKafkaのコンシューマーオフセットをリセットします。

• コンシューマーオフセットがデータ保持期間内にない場合は、例外がスローされ、実行する適切なアクションを決定できます。

• KafkaコネクタがKafkaのコンシューマーオフセットをリセットせず、それを信頼できる情報源として使用する唯一のシナリオは、Snowflakeからのオフセットトークンが NULL である場合です。この場合、コネクタはKafkaによって送信されたオフセットを受け入れ、その後オフセットトークンが更新されます。

記録の処理:

• Kafkaの潜在的なバグによって発生する可能性のある非連続オフセットに対する追加の安全層を確保するために、Snowflakeは最新の処理されたオフセットを追跡するメモリ内変数を維持します。Snowflakeは、現在の行のオフセットが、最後に処理されたオフセットに1を加えた値に等しい場合にのみ行を受け入れます。これにより、インジェスチョンプロセスが継続的かつ正確であることを保証するための追加の保護層が得られます。

例外、障害、クラッシュ復旧への対処:

• 復旧プロセスの一環として、Snowflakeは、チャネルを再開し、最新のコミットされたオフセットトークンを使用してコンシューマーオフセットをリセットすることにより、前に概説したチャネルの開始/再開ロジックを一貫して遵守します。これにより、Snowflakeは、コミットされた最新のオフセットトークンより1つ大きいオフセット値からデータを送信するようKafkaに通知し、データ損失をまったく発生させずに、障害発生時点からインジェスチョンを再開できるようになります。

再試行メカニズムの実装:

• 潜在的な一時的問題に対処するために、Snowflakeには API 呼び出しに再試行メカニズムが組み込まれています。Snowflakeは、成功の可能性を高め、インジェスチョンプロセスに影響を与える断続的な障害のリスクを軽減するために、これらの API 呼び出しを複数回再試行します。

コンシューマーオフセットの提出:

• Snowflakeは、一定の間隔で、最新のコミットされたオフセットトークンを使用してコンシューマーオフセットを提出し、インジェスチョンプロセスが確実にSnowflake内のデータの最新状態と継続的に一致するようにします。

コンバーター¶

Snowpipe Streamingは、以下のような多くのコミュニティベースのコンバーターをサポートしています。

• io.confluent.connect.avro.AvroConverter

• org.apache.kafka.connect.json.JsonConverter

• io.confluent.connect.protobuf.ProtobufConverter

• io.confluent.connect.json.JsonSchemaConverter

• org.apache.kafka.connect.converters.ByteArrayConverter

• org.apache.kafka.connect.storage.StringConverter

その他のコミュニティベースのコンバーターもサポートされるかもしれませんが、検証されていません。SnowflakeコンバーターはSnowpipe Streamingではサポートされていません。

配信不能キュー¶

Snowpipe Streamingを使用するKafkaコネクタは、壊れた記録または障害のために正常に処理できない記録の配信不能キュー（DLQ）をサポートしています。

モニタリングの詳細については、Apache Kafkaの ドキュメント をご参照ください。

スキーマ検出およびスキーマ進化¶

Snowpipe Streamingを使用したKafkaコネクタは、スキーマ検出および進化をサポートしています。Snowflakeのテーブルの構造を自動的に定義して進化させ、Kafkaコネクタによりロードされた新しいSnowpipe Streamingデータの構造をサポートすることができます。Snowpipe Streamingを使用したKafkaコネクタのスキーマ検出と進化を有効にするには、以下のKafkaプロパティを構成します。

• snowflake.ingestion.method

• snowflake.enable.schematization

• schema.registry.url

詳細については、 Snowpipe Streamingを使用したKafkaコネクタのスキーマ検出および進化 をご参照ください。

インジェスチョンレイテンシの推定¶

インジェストレイテンシを見積もるには、 RECORD_METADATA の SnowflakeConnectorPushTime フィールドを使用します。このタイムスタンプは、記録がインジェスト SDK バッファにプッシュされた時点を表します。

RECORD_METADATA フォーマットの詳細については、 Kafkaトピックのテーブルのスキーマ をご参照ください。

請求および使用状況¶

Snowpipe Streamingの課金情報については、 Snowpipe Streamingコスト をご参照ください。

制限事項¶

Snowpipe Streamingの制限については、 制限 をご参照ください。