2026年9月17日/約1分で読めますデータエンジニアリング
データプロデューサーからSnowflakeへのよりシンプルなパスの導入：最大20GB/sをサポートするElastic Channels
本日、Snowpipe StreamingのElastic Channelsの一般提供を発表します。これにより、アプリケーション、デバイス、サービスからSnowflakeテーブルまたはSnowflakeマネージドApache Iceberg™テーブルへデータを直接ストリーミングすることがより簡単になります。データを格納するためだけに存在する取り込みシステムを排除できるため、運用が簡素化され、レイテンシーが短縮し、総保有コストを削減できる可能性があります。
コネクテッドプロダクト、産業用システム、デジタルアプリケーションを運用する組織にとって、このシンプルさは重要です。デバイスは継続的にテレメトリを送信し、アプリケーションはユーザーイベントを生成し、サービスは独立してスケールします。これらのシグナルをアナリティクスやAIで利用できるようにするには、それらを生成するアプリケーションとほぼ同じくらい複雑な取り込みシステムが必要になる場合があります。
Elastic Channelsはその作業を簡素化します。ターゲットテーブルを選択し、行または行のバッチを追加すると、Snowflakeがデータを永続的に保存した時点で（ただし、テーブルでデータがクエリ可能になる前に）永続的な確認応答を受け取ります。SDKは自動的に追加をバッチ処理し、一時的な障害を再試行します。一方、Snowflakeは取り込みのスケーリングとチャネルのライフサイクルを管理します。これにより、Snowflakeにデータを格納するためだけの個別のメッセージバスやステージングパイプラインの必要性が軽減されます。
このよりシンプルなエクスペリエンスは、要求の厳しいワークロードもサポートします。一般提供の時点で、Elastic Channelsはテーブルあたり最大20GB/sをサポートします。ワークロードによっては、データは最短5秒でクエリ可能になります。内部テストではスループットをさらに押し上げ、単一のテーブルへの取り込みで最大40GB/sに達しました。次回の記事では、これらのベンチマーク結果とその測定方法について解説します。また、エンジニアリングに焦点を当てた3番目の記事では、Elastic Channelsの背後にあるアーキテクチャと設計について探求します。
Elastic Channelsの詳細や利用開始については、Elastic Channelsのドキュメントをご覧ください。
お客様の課題から直接的なデータパスへ
Snowflakeは、2つの異なるストリーミングの課題に直面しているお客様とともにElastic Channelsを開発しました。企業向けに報酬およびインセンティブのアプリケーションフレームワークを提供するYuvitalは、パイプラインを簡素化し、ユーザーアナリティクスとクリックストリームイベントをSnowflakeに直接ストリーミングするために、Elastic Channelsをテストしています。産業用DataOpsソフトウェアのリーダーであるHighByteは、何千もの接続をサポートし、産業用およびIoTデータをほぼリアルタイムの可用性でSnowflakeに直接ストリーミングするための、よりシンプルな方法を必要としていました。これらの設計パートナーは、デジタルアプリケーションと産業用システムの両方において、よりシンプルな取り込みの必要性を示しています。
Elastic Channelsは、データをテーブルに移動するためだけに存在する複数のシステムの必要性を減らすことで、取り込みを簡素化します。複数のコンシューマー、共有リプレイ、または保持のためにKafkaのようなメッセージバスをすでに使用している場合は、それを維持したまま、Named Channelsを使用してコピーをSnowflakeにストリーミングできます。
Elastic Channelsを使用すると、お客様はターゲットテーブルを作成し、SDKまたはRESTを介してプロデューサーを接続し、行の追加を開始できます。フリートとトラフィックが増加しても、Snowflakeがサーバー側の取り込みパスを処理します。
“Yuvitalでは、報酬やインセンティブを活用して、組織がメンバーとのエンゲージメントを高め、より健康的な行動を促進できるよう支援しています。メンバーの活動やエンゲージメントイベントの継続的なストリームを生成し、それらをSnowflakeに取り込むことで、メンバーの活動を把握し、エクスペリエンスをパーソナライズして、ビジネス上の意思決定に役立てています。 追記専用イベントをSnowflakeに迅速に取り込む方法としてConfluentを使用していましたが、このようなシンプルな目的にはあまりにも複雑で高価でした。Elastic Channelsを導入したことで、わずか数行のコードを追加するだけで、Node.jsバックエンドから直接イベントを送信できるようになり、拡張が容易になりました。”
Yaron Levi
Chief Architect and Co-Founder, Yuvital
プロデューサーにとってのシンプルさとビジネスにとっての耐久性
Elastic Channelsは、多くの独立したプロデューサーが宛先を共有するワークロード向けに設計されています。
- 1つの共有取り込みパス：プロデューサーは、ターゲットのパイプとテーブルに対して、単一の暗黙的なElastic Channelを介して書き込みます。
- プロデューサーの調整が不要：フリートやトラフィックが変化しても、アプリケーションは取り込みチャネルの割り当て、追跡、再調整を行いません。
- サーバー側の自動スケーリング：プロデューサーの数やトラフィックが変化すると、Snowflakeがワークロードを分散します。
- 永続的な確認応答：Snowflakeが取り込みサービス内で送信されたデータを永続的に保存した後、プロデューサーは確認を受け取ります。
確認応答を受け取った後、プロデューサーは保持しているコピーを解放できます。Snowflakeはデータを永続的に保存しました。処理とクエリの可用性はその後になります。確認応答は、すべての行がダウンストリーム処理を通過したことを意味するものではありません。上記のレイテンシーの数値は、確認応答時間ではなく、取り込みからクエリまでの時間を測定したものです。
プロデューサーは、未確認のイベントを再試行できるように保持しておく必要があります。停止中は、取り込みを一時停止するか、収集を継続する必要がある場合は、既存のアウトボックスまたはプロデューサーローカルの永続バッファを使用できます。クラッシュセーフな保持のためには、責任を受け入れる前にイベントを永続化し、確認応答後にそれらを削除します。SDKのメモリバッファは永続的なスプールではありません。
チームは、すべてのプロデューサーに調整ロジックを組み込むのではなく、Snowflakeをフリートやアプリケーション環境全体における運用データの共有された弾力性のある宛先として扱うことができます。
継続的なデータからビジネス成果へ
多数のプロデューサーが継続的にデータを生成するあらゆる場所で、同じ直接取り込みモデルが適用されます。
- コネクテッドプロダクトとIoT：大規模なフリート全体でデバイスとネットワークの健全性を監視し、テレメトリからプロダクトインテリジェンスまでのパスを短縮し、デバイスごとにチャネルを割り当てることなく新しいプロデューサーをオンボーディングします。
- 産業オペレーション：機械や工場フロアのシグナルをSnowflakeに取り込み、より迅速な異常検知、予知保全（メンテナンス）、および運用の最適化を実現します。
- セキュリティとオブザーバビリティ：分散エージェントからログ、メトリクス、トレース、および行動イベントをストリーミングして、脅威検知とインシデント対応を加速します。
- デジタルアプリケーション：独立してスケールするサービスからクリックストリームやプロダクトイベントをキャプチャし、パーソナライゼーション、実験、およびほぼリアルタイムのお客様アナリティクスをサポートします。
“HighByte Intelligence Hubは、製造業者がエッジ側で機器、センサー、産業システムからのデータを収集してコンテキスト（文脈）を付与し、Snowflake上で利用可能な状態にするのを支援します。Elastic Channelsを活用することで、お客様は個々の工場やデータ生成元（プロデューサー）間でチャネルを個別に調整することなく、共有のインジェスト経路（取り込みパス）を通じて継続的なデータストリームを送信できるため、スループットの向上と管理の簡素化を実現できます。データの欠損が許されない産業（インダストリアル）データにおいて、『永続的な受信確認（Durable acknowledgments）』の仕組みは特に重要です。これにより、お客様はエッジ側でデータを確実にバッファリングし、ネットワーク接続が復旧した際にSnowflakeへとデータを確実に配信することが可能になります。”
Jeffrey Schroeder
Director of Product Management, HighByte
不必要な複雑さのない柔軟なセマンティクス
Snowpipe Streamingは、ユースケースに応じてさまざまなストリーミング機能を提供できます。Elastic Channelsは、順序付けなしで少なくとも1回の配信を提供します。再試行やリプレイによって重複が発生する可能性があるため、重複排除が重要な場合は安定したイベントIDを使用してください。Snowpipe Streamingの既存のNamed Channelsは、チャネルごとの順序付けと正確に1回の配信を必要とするワークロードに対応します。アプリケーションの要件に合った取り込みモデルを選択してください。
Snowflakeデータプラットフォームにネイティブ
データは直接Snowflakeに到着し、組織は一貫したアクセス制御、ガバナンス、変換、アナリティクス、およびAI機能を適用できます。転送レイヤーを削除することで、重複するコピーを減らし、運用データがプラットフォームのガバナンス境界外に置かれる期間を短縮できます。
アプリケーションに合ったクライアントインターフェイスとターゲットテーブルを選択します。
- Python、Java、およびNode.js SDK：イベントが到着したときに追加します。SDKは、時間とサイズのしきい値を使用して追加を自動的にバッチ処理し、一時的な障害を再試行します。アプリケーションは、未確認のまま残るデータの量を制御し、定期的に耐久性の確認を待ちます。
- REST API：軽量サービス、エッジアプリケーション、サーバーレス関数、Webhook、および専用SDKのない言語に最適です。直接のRESTクライアントは、NDJSONのバッチ処理と圧縮を管理します。成功した応答は、永続的な受け入れを確認します。
- 取り込みターゲット：SnowflakeテーブルおよびSnowflakeマネージドIcebergテーブル。
テーブルを選択し、行または行のバッチを追加して、永続的な確認応答を受け取ります。SDKはトランスポートのバッチ処理を処理します。以下のコードサンプルは、単一の追加と確認応答を示しています。
Elastic Channelsと永続的な確認応答は、すべてのAWS、GCP、およびAzureの商用リージョンで一般提供されています。
Elastic Channelsは追加コストなしで使用できます。Snowpipe Streamingは、取り込まれた非圧縮データに基づいて、GBあたり0.0037クレジットで課金されます。詳細については、Snowpipe Streamingドキュメントのコストの理解をご覧ください。
Snowflake CoCoの利用開始
CoCoにElastic Channelsパイプラインの構築を依頼して開始します。
CoCoに「最新のドキュメントhttps://docs.snowflake.com/en/user-guide/snowpipe-streaming/data-load-snowpipe-streaming-overviewを使用して、Snowpipe Streaming Elastic Channelsの利用開始をサポートして」と依頼します。
CoCoは、Snowflakeオブジェクト、認証プロファイル、プロデューサーコード、および検証手順の足場を構築できます。
この最小限のPythonの例は、1つの追加と確認応答を示しています。本番アプリケーションは、未確認のイベントを保持し、すべての行の後ではなく、耐久性チェックポイントで待機します。
from datetime import datetime, timezone
from snowflake.ingest.streaming import StreamingIngestClient
client = StreamingIngestClient.from_table(
client_name="sensor_producer",
db_name="MY_DATABASE",
schema_name="MY_SCHEMA",
table_name="SENSOR_READINGS",
profile_json="profile.json",
)
channel = client.get_elastic_channel()
ack = channel.append_row_with_wait({
"DEVICE_ID": 1,
"READING_TS": datetime.now(timezone.utc),
"TEMPERATURE": 21.5,
})
ack.result() # Data has been durably accepted by Snowflake.
その他のリソース：
- ドキュメントにアクセスしてSnowpipe Streamingの詳細を確認し、Elastic Channelsの利用を開始する
- Python、Java、およびNode.js向けのSnowpipe Streamingサンプルアプリケーションを探索する。
- 公式のクイックスタートガイドを確認する
テーブルを作成し、プロデューサーを接続し、行または行のバッチを追加して、Snowflakeに取り込みのスケーリングとチャネルのライフサイクルを管理させます。
