Snowpipe Streaming¶

Snowpipe Streaming API 対 Snowpipe¶

API はSnowpipeを置き換えるものではなく、Snowpipeを補完することを目的としています。データがファイルに書き込まれるのではなく、行 (例: Apache Kafkaトピック) を介してストリーミングされるストリーミングシナリオでは、Snowpipe Streaming API を使用します。API は、記録を生成または受信する既存のカスタムJavaアプリケーションを含むインジェストワークフローに適合します。 API により、Snowflakeテーブルにデータをロードするためのファイルを作成する必要がなくなり、データが利用可能になると、Snowflakeへのデータストリームの自動継続的なロードが可能になります。

次のテーブルでは、Snowpipe StreamingとSnowpipeの違いについて説明します。

ソフトウェア要件¶

チャネル¶

API は、1つ以上の チャネル を介して行をインジェストします。チャネルは、データをテーブルにロードするためのSnowflakeへの論理的な名前付きストリーミング接続を表します。単一のチャネルは、Snowflakeの1つのテーブルに必ずマップされます。ただし、同じテーブルを複数のチャネルがポイントすることはできます。クライアント SDK は複数のチャネルを複数のテーブルに開くことができますが、 SDK はアカウント間でチャネルを開くことはできません。行の順序とそれに対応するオフセットトークンはチャネル内で保持されますが、同じテーブルをポイントするチャネル間では保持されません。

チャネルは、クライアントが活発にデータを挿入していると長い間維持され、オフセットトークン情報が保持されるため再利用されるべきです。チャネル内のデータはデフォルトで1秒ごとに自動的にフラッシュされ、閉じる必要はありません。詳細については、 遅延 をご参照ください。

チャネルと関連するオフセットメタデータが不要になった場合、 DropChannelRequest API を使用して、チャネルを永久に削除することができます。チャネルを削除するには2つの方法があります:

• クロージングでチャネルを削除する。チャネル内のデータは、チャネルが削除される前に自動的にフラッシュされます。

• やみくもにチャネルを削除する。やみくもにチャネルを削除すると、保留中のデータがすべて破棄され、すでにオープンされているチャネルが無効になる可能性があるため、これは推奨されません。

SHOW CHANNELS コマンドを実行すると、アクセス権限のあるチャンネルをリストできます。詳細については、 SHOW CHANNELS をご参照ください。

オフセットトークン¶

オフセットトークン は、チャネルごとにインジェスチョンの進行状況を追跡するために、クライアントが insertRow （単一行）または insertRows （行のセット）メソッドリクエストに含めることができる文字列です。トークンはチャネルの作成時に NULL に初期化され、提供されたオフセットトークンを持つ行が非同期プロセスを通じてSnowflakeにコミットされるときに更新されます。クライアントは定期的に getLatestCommittedOffsetToken メソッドリクエストを作成して、特定のチャネルに対してコミットされた最新のオフセットトークンを取得し、それを使用してインジェスチョンの進行状況を判断できます。このトークンは、Snowflakeが重複除外を実行するためには使用 しない ことに注意してください。ただし、クライアントはこのトークンにより、カスタムコードを使用して重複除外を実行することはできます。

クライアントがチャネルを再度開くと、最新の永続化されたオフセットトークンが返されます。クライアントは、トークンを使用してデータソース内の位置をリセットし、同じデータを2回送信することを回避できます。チャネルの再開イベントが発生すると、Snowflakeにバッファーされたデータはコミットを回避するために破棄されることに注意してください。

最新のコミットされたオフセットトークンを使用して、次の一般的な使用例を実行できます。

• インジェスチョンの進行状況の追跡

• 特定のオフセットがコミットされているかどうかを、最後にコミットされたオフセットトークンと比較することによって確認します。

• ソースオフセットの提出と、すでにコミットされたデータのパージ

• 重複排除の有効化と、データを1回だけ配信することの保証

たとえば、Kafkaコネクタはトピックから <パーティション>:<オフセット> などのオフセットトークンを読み取ることができます。または、パーティションがチャネル名にエンコードされている場合は、単に <オフセット> を読み取ることができます。次のシナリオを検討してください。

1. Kafkaコネクタがオンラインになり、Kafkaトピック T の Partition 1 に対応するチャネルが T:P1 というチャネル名で開きます。

2. コネクタは、Kafkaパーティションからの記録の読み取りを開始します。

3. コネクタは API を呼び出し、記録に関連付けられたオフセットをオフセットトークンとして使用して insertRows メソッドリクエストを作成します。

   たとえば、オフセットトークンは 10 で、Kafkaパーティションの10番目の記録を参照します。

4. コネクタは定期的に getLatestCommittedOffsetToken メソッドリクエストを実行し、インジェストの進行状況を判断します。

Kafkaコネクタがクラッシュした場合は、次のプロシージャを完了して、Kafkaパーティションの正しいオフセットから記録の読み取りを再開できます。

1. Kafkaコネクタがオンラインに戻り、以前と同じ名前を使用してチャネルを再度開きます。

2. コネクタは API を呼び出し、 getLatestCommittedOffsetToken メソッドリクエストを作成して、パーティションの最新のコミット済みオフセットを取得します。

   たとえば、最新の永続オフセットトークンが 20 であるとします。

3. コネクタは、Kafka読み取り APIs を使用して、オフセット+1 (この例では 21)に対応するカーソルをリセットします。

4. コネクタは記録の読み取りを再開します。読み取りカーソルが正常に再配置されると、重複するデータは取得されません。

別の例では、アプリケーションはディレクトリからログを読み込み、Snowpipe Streaming Client SDK を使用してこれらのログをSnowflakeにエクスポートします。次を実行するログエクスポートアプリケーションをビルドできます。

1. ログディレクトリ内のファイルをリストする。

   ログフレームワークが生成するログファイルは辞書順に並べることができ、新しいログファイルはこの順序の最後に配置されるとします。

2. ログファイルを1行ずつ読み取り、 API を呼び出して、ログファイル名と行数またはバイト位置に対応するオフセットトークンを使用して insertRows メソッドリクエストを作成する。

   たとえば、オフセットトークンは messages_1.log:20 のようになります。ここで、 messages_1.log はログファイルの名前で、 20 は行番号です。

アプリケーションがクラッシュしたり、再起動が必要になったりすると、 API が呼び出され、最後にエクスポートされたログファイルと行に該当するオフセットトークンを取得する getLatestCommittedOffsetToken メソッドリクエストが作成されます。例を続けると、これは messages_1.log:20 になります。その後、アプリケーションは messages_1.log を開き、行 21 を検索して、同じログ行が2回インジェストされないようにします。

1回限りの配信のベストプラクティス¶

正確に1回限りの配信を実現することは困難な場合があり、カスタムコードでは次の原則に従うことが重要です。

• 例外、障害、またはクラッシュから確実かつ適切に復旧するには、常にチャネルを再開し、最新のコミットされたオフセットトークンを使用してインジェスチョンを再起動する必要があります。

• アプリケーションは独自のオフセットを維持できますが、Snowflakeによって提供される最新のコミットされたオフセットトークンを信頼できる情報源として使用し、それに応じて独自のオフセットをリセットすることが重要です。

• 独自のオフセットを信頼できる情報源として扱う必要がある唯一の例は、Snowflakeからのオフセットトークンが NULL に設定またはリセットされる場合です。NULL オフセットトークンは通常、次のいずれかを意味します。

  • これは新しいチャネルであるため、オフセットトークンが設定されていない。

  • ターゲットテーブルがドロップされて再作成されたため、チャネルが新規とみなされます。

  • 30日間、チャネルを介したインジェスチョンアクティビティがなかったため、チャネルは自動的にクリーンアップされ、オフセットトークン情報は失われました。

• 必要に応じて、最新のコミットされたオフセットトークンに基づいてすでにコミットされているソースデータを定期的にパージし、独自のオフセットを提出することができます。

• Snowpipe Streamingチャネルがアクティブなときにテーブルスキーマが変更された場合、チャネルを再度開く必要があります。Snowflake Kafkaコネクタはこのシナリオを自動的に処理しますが、Snowflake Ingest SDK を直接使用する場合は、自分でチャネルを再オープンする必要があります。

Snowpipe Streamingを使用したKafkaコネクタが1回限りの配信を実現する方法の詳細については、 必ず1回のセマンティクス をご参照ください。

遅延¶

Snowpipe Streamingは、チャネル内のデータを1秒ごとに自動的にフラッシュします。データをフラッシュするためにチャネルを閉じる必要はありません。

Snowflake Ingest SDK バージョン 2.0.4以降では、オプション max_client_lag を使用して遅延を設定できます。デフォルトのオプションは1秒です。最大遅延時間は10分まで設定できます。詳細については、 MAX_CLIENT_LAG および Snowpipe Streamingの推奨レイテンシ構成 をご参照ください。

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

最適化されたファイルへの移行¶

API は、チャネルからの行をクラウドストレージのBLOBに書き込み、ターゲットテーブルにコミットします。最初に、ターゲットテーブルに書き込まれたストリームデータは、仮の中間ファイル形式で格納されます。このステージでは、テーブルは「混合テーブル」と見なされます。これは、ネイティブファイルと中間ファイルが混ざった状態でパーティションされたデータが格納されているためです。自動化されたバックグラウンドプロセスにより、必要に応じて、アクティブな中間ファイルから、クエリおよび DML 操作用に最適化されたネイティブファイルにデータが移行されます。

複製¶

Snowpipe Streamingは、Snowpipe Streamingとそれに関連するチャネルオフセットによって入力されたSnowflakeテーブルの 複製とフェールオーバー を、異なる リージョン のソースアカウントとターゲットアカウント間、また クラウドプラットフォーム 間でサポートしています。

詳細については、 複製およびSnowpipe Streaming をご参照ください。

挿入のみの操作¶

API は現在、行の挿入に限られています。データを変更、削除、または結合するには、「生」記録を1つ以上のステージングされたテーブルに書き込みます。 継続的なデータパイプライン を使用してデータをマージ、結合、または変換し、変更されたデータを宛先のレポートテーブルに挿入します。

クラスおよびインターフェイス¶

クラスとインターフェイスのドキュメントについては、 Snowflake Ingest SDK API をご参照ください。

サポートされているJavaデータ型¶

次のテーブルは、Snowflake列へのインジェスチョンでサポートされているJavaデータ型をまとめたものです。

必要なアクセス権限¶

Snowpipe Streaming API を呼び出すには、次の権限を持つロールが必要です。

制限事項¶

Snowpipe Streamingは、データ暗号化に256ビット AES キーの使用のみをサポートします。

以下のオブジェクトや型はサポートされて いません 。

• GEOGRAPHY および GEOMETRY データ型

• 列の照合順序設定

• TRANSIENT または TEMPORARY テーブル

• 以下のいずれかの列設定を持つテーブル:

  • AUTOINCREMENT または IDENTITY

  • NULL ではないデフォルトの列値

• テーブルあたりのチャネル数の合計は10kを超えることはできません。必要に応じてチャネルを再利用することをお勧めします。1テーブルあたり10kチャネル以上を開く必要がある場合は、 Snowflakeサポート までお問い合わせください。