インフラストラクチャの構築不要で本格的なストリーミングを評価
Snowpipe Streamingが100万以上のTPSに耐えられるかどうかを確認したいチームにとって、確実な答えを得るための最短ルートが必要になります。大規模なストリーミングには、特有の難しさがあると言われています。S3バケット、EKSクラスター、Kafkaトピック、IAMポリシーのセットアップに加え、それぞれ独自のバックログを抱える3つのプラットフォームチームからのセキュリティレビューと承認が必要になるため、チームが後回しにしがちなワークロードです。このようなセットアップでは、テストを開始する前に、追加のインフラストラクチャのプロビジョニングと調整が必要になる場合があります。これらのパイプラインが大規模に機能し、本番環境のボリュームを処理できることを証明すること自体が、通常は1つの独立したプロジェクトになります。
今回のデモ環境では、この評価のセットアップと実行を半日で完了できました。Snowpipe Streamingの高性能アーキテクチャを使用すると、すべてSnowflakeのインフラストラクチャ内で、100万TPS以上の速度でApache Iceberg™形式にデータをストリーミングできます。Snowparkコンテナサービス（SPCS）で実行されるコンテナが負荷を生成し、SDKを介してストリームし、SnowflakeマネージドのIcebergテーブルに格納します。これにより、データが到着すると同時にロールベースのアクセス制御（RBAC）、リネージ、マスキングがすべて適用され、Snowflakeで迅速にクエリを実行できるようになります。テスト環境では、デモの立ち上げ、エンドツーエンドのテスト、スループットの測定、そして解体を、すべて半日で完了しました。
本記事では、このGitHubリポジトリを使用して、Icebergへのエンドツーエンドのストリーミングデモを実際に構築する方法を説明します。この例はPythonで記述されていますが、SDKのオプション（Java、Python、ノード、REST）をテストできるため、実際の成果を示してビジネスに自信を持って推奨できます。
アーキテクチャ図：パターン
すべてがSnowflakeのインフラストラクチャ内で実行されるデモを構築したいと考えました。Snowparkコンテナサービス（SPCS）で実行される複数のDockerコンテナが、目標とする1秒あたりのトランザクション数で合成テストデータを生成し、Snowpipe Streaming SDKを使用してストリーミングします。データはSnowflakeマネージドのIcebergテーブルに格納されます。これはオープンフォーマットであり、格納後数秒以内にクエリ可能で、Snowflake内でガバナンスが適用されます。認証は簡単です。SPCSは、短期間有効なOAuthトークンをコンテナに挿入し、Streaming SDKにそれを使用するように指示します。シークレットを管理する必要はありません。
負荷生成にSPCSを使用する理由
誤解のないように言えば、Snowpipe StreamingにSPCSは必須ではありません。SDKは、ノートパソコン、EC2インスタンス、Kubernetesポッドなど、どこでも実行できます。
SPCSを使用すると、Snowflakeがすでに管理しているインフラストラクチャ上でサービスが完全に実行されるため、テストのために新しくプロビジョニングするものや追加の承認を管理する必要がなくなり、テストをより迅速に開始できます。
Icebergへの直接格納
ターゲットは、SnowflakeマネージドのIcebergテーブルです。行はSDKを介してストリーミングされ、SnowflakeによってマネージドされるIcebergメタデータとともにParquetとして格納されます。必要に応じて後から独自のS3外部ボリュームを指定することも、評価後にマネージドのままにしておくこともできます。どちらの場合でもコードは変わりません。
評価においては、セットアップの手間が省けることを意味します。Snowflakeマネージドのストレージ上のテーブルを使用すると、通常は最初に構成する必要がある外部ボリューム、ストレージ統合、クラウドストレージの権限設定を省略できます。
本番環境に移行する準備が整ったら、3つの明確なパスがあります。
- 外部ボリュームを使用するSnowflakeマネージドのIceberg：ParquetファイルはS3バケットに保存されます。外部エンジンは、メタデータにHorizon RESTカタログを使用して、S3から直接Parquetファイルを読み取ります。
- Snowflakeストレージ上のSnowflakeマネージドのIceberg：ファイルはSnowflakeの内部ストレージに保存されます。このオプションでは、外部ボリュームと関連するクラウドストレージの構成を回避できます。今後の互換性のためにIcebergの機能を活用します。
- ネイティブSnowflakeテーブル：この構成では、個別にマネージドされる外部ストレージインフラストラクチャは必要ありません。
デモ：実際の動作
デモスクリプトを実行すると、SPCSサービスが開始され、プロデューサーが負荷の生成を開始し、コンシューマーがそれをIcebergテーブルにストリームするのを確認できます。
Snowsightに切り替えると、ここでも行数が増加していることを確認できます。数百万の行が、格納されると同時にクエリ可能になります。
コード：ご自身での構築
この仕組みを理解しやすくするため、視覚的にわかりやすい疑似コードを以下に示します。コード全体はリポジトリにあり、以下にいくつかの代替デモへのリンクも掲載しています。
まずは、SnowflakeマネージドのIcebergテーブルとパイプから始めます：
-- 1. The target table
CREATE OR REPLACE ICEBERG TABLE events (
event_id STRING,
event_ts TIMESTAMP_NTZ,
payload VARIANT
)
CATALOG = 'SNOWFLAKE'
BASE_LOCATION = 'events/'
ICEBERG_VERSION = 3;
-- 2. The streaming pipe (extracts typed fields from the SDK's VARIANT payload)
CREATE OR REPLACE PIPE events_pipe
AS COPY INTO events (event_id, event_ts, payload)
FROM (
SELECT $1:event_id::STRING,
$1:event_ts::TIMESTAMP_NTZ,
$1:payload::VARIANT
FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);
Snowpipe Streamingの高性能アーキテクチャはv2とv3の両方をサポートしていますが、パラメータを省略するとテーブルはデフォルトでv2になります。
次に、コンシューマーです：
from snowflake.ingest.streaming import StreamingIngestClient
# SPCS injects credentials automatically - no keys, no secrets
props = {
"account": "YOUR_ACCOUNT",
"user": "YOUR_USER",
"role": "STREAMING_SERVICE_ROLE",
"url": "https://YOUR_ACCOUNT.snowflakecomputing.com",
"authorization_type": "SPCS",
"spcs_token_path": "/snowflake/session/token",
}
# Connect to the pipe
client = StreamingIngestClient(
client_name="my_consumer",
db_name="STREAMING_DEMO",
schema_name="PUBLIC",
pipe_name="events_pipe",
properties=props,
)
# Open a channel and stream rows
channel, status = client.open_channel("ch_1")
for i, row in enumerate(generate_load()):
channel.append_row(row, offset_token=str(i))
# Close cleanly
channel.close()
client.close()
その後、短いサービス仕様でコンシューマーをSPCSに展開します。そして、データが取り込まれるのを確認します：
SELECT
COUNT(*) AS rows_landed,
MAX(event_ts) AS latest_event
FROM
streaming_demo.public.events;Java、ノード、RESTインターフェイスはすべて同じ形式に従います。つまり、チャネルを開き、行を挿入して、フラッシュします。リポジトリにはそれぞれ実際に動作する例があるため、ご自身のスタックに合ったものをテストできます。
ご自身で試すには、リポジトリをご確認ください。
ただし、Streaming SDKはHTTPS経由で別の取り込みエンドポイントを使用するため、Snowflake内で実行している場合でも、*.snowflakecomputing.comへのエグレスを許可する外部アクセス統合を付与する必要があることに注意してください。
ベストプラクティス
Snowpipe Streamingによる自動バッチ処理
Snowpipe Streamingは、行ごとにファイルを書き込むことはありません。SDKはappendRowsを呼び出すとすぐに行を送信しますが、サーバー側のバッファリング層がそれらを吸収して独自のタイミングでコミットを決定し、バックグラウンドでバッチ処理を行うため、行ごとにファイルが送信されることはありません。
さらに、Snowflakeが唯一のライターであるSnowflakeマネージドのIcebergテーブルに対して、Snowflakeはバックグラウンドで自動コンパクションを実行します：小さなParquetファイルは大きなファイルにマージされ、小さなマニフェストもコンパクションされます。スケジュールを設定したり実行したりする必要はなく、通常の操作に組み込まれています。
ユーザーができること：
- 呼び出し前のクライアント側での行のバッチ処理：
appendRows：行を1つずつ送信すると、依然として行ごとに1回のラウンドトリップが発生する。これらを1回の呼び出しにバッチ処理することで、そのオーバーヘッドを軽減できる。これは、リクエストごとにより多くのデータを圧縮して送信するというSnowflake独自のガイダンスの背後にある原則と同じである - チャネルの長期間の維持：マイクロバッチごとに開閉するのではなく、ソースパーティションごとにチャネルを1回開き、ジョブの実行中はそのチャネルを開いたままにする。これはSnowflakeがドキュメントで推奨するベストプラクティス（Snowpipe Streaming向け）であり、バックグラウンドですでに実行されているバッチ処理と競合することなくオーバーヘッドを削減する
- 設定対象：
TARGET_FILE_SIZE（テーブル）：これはSDKとは独立したテーブルレベルのプロパティであり、データがどのように取り込まれたかにかかわらず、新規書き込みとバックグラウンドコンパクションの両方でターゲットとするサイズをSnowflakeに指示する - スループットの活用： 持続的なTPSが高いほど、サーバー側のバッファがより速く満たされて自動的にフラッシュされる。これにより、特定のファイルサイズにおけるスループットとレイテンシーのトレードオフが効果的にもたらされる
今後の展望：Iceberg v4
Snowpipe Streamingと自動コンパクションは物理的なスモールファイルの問題に対処しますが、頻繁なコミットは依然としてIcebergのメタデータツリーにオーバーヘッドを生じさせる可能性があります。現在、各コミットは、たとえ1つの小さなファイルしか含まれていない場合でも、新しいメタデータJSON、マニフェストリスト、およびマニフェストを書き込みます。数秒ごとにコミットするワークロードの場合、その繰り返されるメタデータ処理がボトルネックになる可能性があります。
Iceberg v4の提案では、その書き込み増幅を減らすために適応型メタデータツリー（Adaptive Metadata Tree）を導入しています。小さなコミットごとに新しいマニフェストを作成する代わりに、ルートマニフェストがそれらのコミットをインライン化できます。1回のParquet書き込みとアトミックなポインタの交換が現在のチェーンを置き換え、コミットごとのメタデータI/Oがマニフェスト数に応じて増加するのではなく、一定に保たれます。高頻度のストリーミングワークロードの場合、これによりストリーミングレイテンシーが大幅に向上する可能性があります。
ストリーミングは、Iceberg v4の主要なシナリオの1つです：数秒ごとにコミットするワークロードは、メタデータの書き込み増幅を直接露呈させるため、適応型メタデータツリーの利点が特に関連してきます。v4の仕様はまだ進化していますが、Icebergストリーミングがロードマップにある場合は注目する価値があります。Snowflakeは、このフォーマットの未来を形作るために積極的に参加していることを誇りに思っています。
アーキテクチャへの影響
本記事では、IcebergへのSnowpipe Streamingがいかに効果的であるかを示しました。本格的なメディアや通信プラットフォームが稼働するような負荷において、このテストセットアップではS3バケット、EKSクラスター、Kafkaトピック、IAMロールを必要とせず、ワークロードはSnowflake内で実行され、評価後に破棄することができました。
通常、チームとストリーミングテストの間に存在するインフラストラクチャ（バケット、クラスター、認証、承認など）は前提条件ではありません。「これが要件を満たしているか」という問いに迅速に答えることができます。
大規模なメディアや通信プラットフォームのサイト信頼性エンジニアリング（SRE）またはデータエンジニアリングの担当者であり、Icebergストリーミングがロードマップにある場合は、調達サイクルではなくテストから始めることをお勧めします。ぜひリポジトリをクローンし、SPCSに展開して、その動作をご確認ください。
- GitHubリポジトリから今すぐお試しいただけます。
- さらに詳しく知るには、Snowpipe Streamingの高性能アーキテクチャに関するドキュメントをご確認ください。
