Snowpipe StreamingとSnowflake CoCoによるほぼリアルタイムのパイプラインのシンプルかつ低コストなリリース

金融サービス業界のデータエンジニアは、他の業界とは異なるプレッシャーにさらされています。市場の動きは急速です。リスクモデルには、数分前の状況ではなく、現在の状況を反映させる必要があります。不正検出は、トランザクションが完了する前にシグナルが到達して初めて機能します。リアルタイムダッシュボードには、リアルタイムのデータを表示する必要があります。

問題は、ほとんどのデータインフラストラクチャが、そのような緊急性を考慮して構築されていないことです。スケジュールに従って実行されるバッチパイプラインは、多くのワークロードで適切に機能しますが、金融のユースケースでは許容できないレイテンシーが発生します。この課題を認識しているチームは、ストリーミングアーキテクチャの構築に何週間も費やすことがよくありますが、実際のパイプラインロジックを記述するずっと前に、セットアップの複雑さによって作業が遅れてしまいます。データエンジニアがインフラストラクチャの基盤構築に費やす時間は、重要な成果の達成に費やされない時間となります。そのようなオーバーヘッドは蓄積していきます。

Snowpipe StreamingのハイパフォーマンスアーキテクチャとSnowflake CoCoは、まさにこの問題を解決するために設計されています。

Snowflakeへのほぼリアルタイムなデータ取り込み

Snowpipe Streamingのハイパフォーマンスアーキテクチャは、データエンジニアがPython SDKを使用してアプリケーションコードからSnowflakeに直接行を書き込むことができる、直接取り込みAPIです。行は10秒未満で取り込まれ、スループットは最大10GB/秒に達し、ほぼ即時にクエリを実行できます。ステージングのステップやCOPY INTOコマンドは不要であり、維持すべきファイル管理レイヤーもありません。

これは、特に金融のワークロードにおいて重要です。市場イベント、オーダーブックの更新、トランザクション記録はすべて、迅速に移動し、アナリスト、モデル、リスクシステムがアクションを起こせる場所に取り込まれる必要があります。Snowpipe StreamingはSnowflake内でネイティブに実行されるため、書き込まれる実質的にすべての行は、組織で導入済みのガバナンス、アクセス制御、リネージを継承します。管理すべきステージングレイヤーはなく、分析プラットフォームと並行して運用する個別のシステムもありません。

エンジニアが実際に時間を浪費する領域

API自体は難しいものではありません。実際に使用できる状態にするまでのプロセスが困難なのです。

1つの行が取り込まれる前に、チームは認証のセットアップ、適切なSnowflakeオブジェクトと権限のプロビジョニング、開発環境の構成を行い、実行中のパイプラインが正常であることを検証する仕組みを構築する必要があります。これらはどれも理不尽なことではありませんが、Snowpipe Streamingの使用を決定してから実際に使用するまでの間に存在する、膨大な作業です。これは、概念実証（PoC）のパイプラインが本番環境に到達する前に停滞する主な理由の1つでもあります。

Snowpipe Streaming向けのCoCoスキル

Snowflake CoCoは、Snowsight、CLI、Desktopで利用できるデータネイティブなAIコーディングエージェントです。ユーザーのSnowflake環境、その中のオブジェクト、Snowpipe Streamingなどのサービスの構成方法を理解します。何かが実行される前に、レビューと承認のためのプランが提示されるため、ユーザーは実行内容を常に制御できます。

利用開始に役立つスキル

GitHubのSSv2-AI-Webinarリポジトリでは、この実践に役立つ2つのスキルを利用できます。
 

ssv2-quickstartスキルは、セットアップシーケンス全体を実行します。必要なSnowflakeオブジェクトの作成、認証の処理、Python取り込みスクリプトの記述を行い、ライブのStreamlitモニタリングダッシュボードを構築するため、パイプラインが開始された瞬間にデータの流れを確認できます。通常であればドキュメントと構成の間を行き来して何時間もかかる作業を、CoCoは単一のガイド付きセッションで処理します。プロセス全体はわずか数分で完了します。

利用開始のプロンプト例：ssv2 quickstartまたはsnowpipe streaming

ssv2-AI-webinarスキルは、その基盤の上に構築されます。このスキルは、実行中のSnowpipe StreamingパイプラインをSnowflake内のCortex AI関数に接続し、ほぼリアルタイムのイベント分類、エンティティ抽出、エンリッチメントを行う方法を示します。数秒で取り込まれた金融データを、他の場所に移動することなく、同じプラットフォーム内で分析してアクションを起こせます。

プロンプト例：ssv2 ai webinarまたはssv2 webinar demo

携帯電話基地局のエンドツーエンドの予測メンテナンスの構築

SDK Examplesで提供されているスキルを使用することで、携帯電話基地局の予測メンテナンスのエンドツーエンドのユースケースを構築できます。このワークフローは、実行中のSnowpipe StreamingパイプラインをSnowflake内のCortex AI関数と統合し、イベント分類やエンティティ抽出などの、ほぼリアルタイムのデータ分析をプラットフォーム内で直接可能にする方法を示します。

このスキルは、ローカルのKafkaクラスター、Kafkaトピックにデータを供給するプロデューサーアプリケーション、Snowpipe StreamingおよびSnowflake AI関数に書き込むカスタムKafkaコンシューマーをセットアップし、基地局のドロップ率などの主要なKPIを予測します。

この例では、Snowpipe Streaming SDKを使用してカスタムKafkaコンシューマーを構築するためのベストプラクティスや、再試行およびエラー処理ロジックに関するベストプラクティスも紹介しています。

プロンプトの例：snowpipe streaming kafka

ワークフロー実行中のCoCoの動作

パイプラインの稼働後は、実際の環境に基づいた質問をCoCoに投げかけることができます。スループットが想定より低い場合は、その理由を質問できます。データの変化に伴ってイベントスキーマを拡張する必要がある場合は、要件を説明することで、CoCoが特定のテーブルに合わせたSQLを生成します。アップストリームから不正な形式の行が送信された場合、CoCoはエラー処理の追加やその動作テストを支援します。

CoCoはセッション開始時からのコンテキストを保持しているため、問題が発生するたびに設定を説明し直す必要はありません。ドキュメント、CLIツール、Snowflakeコンソールを行き来することなく、ワークフロー全体を1か所で完結できます。