ブロックチェーン・データ・ストリーミングとは何ですか？

わかりやすい説明

システムからデータを取得するには、基本的に2つの方法があります。データを要求する（プル）方法と、データが送信されてくる（プッシュ）方法です。 ブロックチェーンの歴史の大部分において、開発者は「プル」モデルを採用してきました。アプリケーションがノードにRPCリクエストを送信すると、ノードからレスポンスが返され、アプリケーションが次にリクエストを送信するタイミングを決定します。これが「ポーリング」パターンです。実装は簡単ですが、特にアプリケーションがブロックチェーンの最新状態と同期を維持する必要がある場合、非効率的です。

ストリーミングはこのモデルを逆転させます。アプリケーションが1日に何千回も「最新のブロックで何が起きたか」と問い合わせる代わりに、データパイプラインを一度設定するだけで、ストリーミングサービスが新しいブロックが生成されるたびに、そのデータをアプリケーションやデータベースに配信します。これにより、アプリケーションは個別のデータスナップショットを要求する存在ではなく、継続的なデータフィードの消費者となります。

この例えは、ニュースサイトを5分おきに更新し続けることと、ニュースが配信された瞬間に通知が届くプッシュ通知を購読することの違いに例えられます。どちらも最終的には同じ情報を得られますが、プッシュ通知の方がより迅速で、無駄が少なく、信頼性が高いのです。

ブロックチェーンによるデータストリーミングの仕組み

ストリーミングサービスは、ブロックチェーンのノードと、アプリケーションやデータインフラストラクチャとの間に位置します。データ取り込み側では、このサービスはサポートされているネットワーク上のノードに接続し、新しいブロックが確定するたびにそれを処理します。サービスは、ヘッダー、トランザクション、トランザクションレシート（イベントログを含む）、およびオプションで実行トレースを含む生のブロックデータを抽出します。この生データは、そのブロック内で発生したすべての事象の完全な記録となります。

処理の面では、ストリーミングサービスはユーザーが設定したフィルタや変換を適用します。たとえば、特定の契約群からのERC-20転送イベントのみに関心がある場合を考えてみましょう。JavaScriptやその他のサポートされている言語で記述されたフィルタが各ブロックのデータに対して実行され、設定した条件に一致するレコードのみを通過させます。このサーバーサイドでのフィルタリングは、ブロック全体ではなく、実際に必要なデータのみを宛先が受信・保存することになるため、効率性の観点から極めて重要です。

配信側では、フィルタリングされたデータが、設定された送信先に送信されます。送信先としては、アプリケーションがリスニングしている Webhook URL、PostgreSQL データベース、Snowflake データウェアハウス、Amazon S3 バケット、Azure Blob Storage、またはその他のサポートされているエンドポイントなどが考えられます。ストリーミングサービスは、失敗時の再試行、重複排除、正しい順序の保証といった配信保証を処理するため、送信先にはブロックチェーンデータが途切れることなく、時系列順に配信されます。

×

Quicknode

□

ストリーミングと従来のETLアプローチの比較

従来のブロックチェーンETL（抽出、変換、ロード）パイプラインでは、開発者が大規模なインフラを構築・維持する必要があります。 典型的なカスタムETL構成には、スケジュールに従ってRPCエンドポイントを呼び出して新しいブロックを取得するポーリングワーカー、生データをデコードおよび変換する処理層、レコードを挿入するデータベースライター、再試行や障害を管理するエラーハンドラー、放棄されたフォークからのデータを特定してロールバックする再編成検出器、そしてこれらのコンポーネントのいずれかに障害が発生した際にアラートを発する監視機能が含まれます。それぞれの要素が複雑さを増し、それぞれが潜在的な障害点となります。

ストリーミングサービスは、これらすべてを単一の管理対象パイプラインに統合します。データの抽出、フィルタリング、変換、配信、再試行ロジック、順序付け、および再編成の処理はすべて、プロバイダーのインフラストラクチャ上で行われます。ユーザーの責任は、必要なデータ、送信先、およびデータの整形方法を設定することに限定されます。これにより、ブロックチェーンデータインフラストラクチャの構築と維持に必要なエンジニアリングの労力が劇的に削減されます。

パフォーマンスの差も甚大です。ポーリングベースのパイプラインでは、その性質上、ポーリング間隔に等しいレイテンシが発生します。ワーカーが5秒ごとに新しいブロックの有無を確認する場合、すべてが完璧に動作していても、チェーンの先端から最大5秒遅れてしまう可能性があります。エラー、レート制限、またはワーカーの再起動などによるポーリングの失敗は、この遅れをさらに拡大させます。 ストリーミングでは、データが利用可能になり次第プッシュされるため、ポーリングによるレイテンシが排除されます。数百ミリ秒ごとにブロックが生成される高スループットのチェーンでは、レイテンシに敏感なアプリケーションにとって、この違いは大きな意味を持ちます。

×

Quicknode

□

プロダクション・ストリーミングの主な特徴

本番環境向けのストリーミングには、単純なWebSocketサブスクリプションとは異なるいくつかの特徴があります。配信保証機能により、たとえ接続が一時的に切断されたり、送信先で一時的な障害が発生したりした場合でも、各ブロックのデータが確実に1回だけ送信先に到達することが保証されます。ストリーミングサービスは、配信が確認されるまでデータをバッファに保持し、再送信を繰り返すため、データに欠落が生じることはありません。

「ファイナリティ順の配信」とは、データがブロックチェーンが正規とみなす順序で到着することを意味します。ファイナリティ時間が変動するチェーンでは、ストリーミングサービスはブロックが十分に確認されるまで配信を待機し、リオーガナイゼーションによって後で無効化される可能性のあるデータをアプリケーションが処理することを防ぎます。リオーガナイゼーションが発生した場合、サービスはフォークを検出し、どのブロックがもはや正規ではないかを特定した上で、修正ペイロードを送信し、宛先がそれに応じて更新できるようにします。

履歴データのバックフィル機能により、新しいブロックだけでなく、過去のデータに対しても同じストリーミングパイプラインを使用できるようになります。履歴データ用に別のETLプロセスを構築し、リアルタイムデータ用に別のプロセスを構築する代わりに、過去のブロックを起点とする単一のストリームを設定し、履歴データを順次処理させていき、チェーンの先端に追いついた時点でシームレスにリアルタイム処理へ移行させます。この統合されたパイプラインにより、アーキテクチャが簡素化され、履歴データとリアルタイムデータ間の一貫性が確保されます。

サーバーサイドでのフィルタリングと変換により、宛先に届くデータ量が削減され、帯域幅コスト、ストレージコスト、および処理オーバーヘッドが低減されます。フィルタでは、アドレス、イベントシグネチャ、関数セレクタ、トランザクション値、あるいはブロックデータのその他のプロパティに基づいて照合を行うことができます。変換では、ペイロードの再構成、ABIエンコードされたデータのデコード、派生値の計算、およびデータベーステーブルに一致するスキーマでのレコード出力を行うことができます。

×

Quicknode

□

Quicknode Streamsによるブロックチェーン・データ・ストリーミングの実現方法

QuicknodeStreamsは、 80以上のブロックチェーンにわたるリアルタイムデータおよび履歴データを提供する、専用に設計されたブロックチェーンデータストリーミングおよびETLサービスです 。フィルタリング、変換、配信保証機能が組み込まれています。Streamsは、複数のデータセットタイプ（ブロック、トランザクション、レシート、イベントログ、トレース）および複数の送信先（Webhook、PostgreSQL、Snowflake、Amazon S3、Azure Storageなど）に対応しています。

Streamsは、ファイナリティ順でデータを処理し、「正確に1回」の配信を保証します。また、修正ペイロードを送信することでチェーンの再編成を自動的に処理し、履歴データのバックフィル時に最適なスループットを実現するための、設定可能なバッチ処理と圧縮をサポートしています。JavaScriptフィルターはQuicknodeのインフラ上で実行されるため、データが宛先に到達する前に、イベントのデコード、パターンの照合、外部キーバリューストアへの参照、出力ペイロードの整形を行うことができます。 ワンクリックのバックフィルテンプレートは、20以上のチェーンにまたがる一般的なデータセット向けに事前設定済みのパイプラインを提供し、開始前に透明性の高いコストと完了時間の見積もりが表示されます。

さらに高度なワークフローを実現するため、StreamsはQuicknode Functionsと連携し、ストリーミングデータを活用したサーバーレスな自動化を可能にします。Functionsでは、ストリーミングイベントに応じて、レコードに追加のオンチェーンデータを付加したり、外部APIを呼び出したり、通知をトリガーしたり、任意のビジネスロジックを実行したりすることができます。StreamsとFunctionsを組み合わせることで、カスタムETLインフラストラクチャに代わる、管理されたスケーラブルなソリューションを備えた、包括的なブロックチェーンデータプラットフォームが提供されます。

ストリーミングとポーリングの違いは何ですか？

ポーリングとストリーミングは、アプリケーションがオンチェーンのアクティビティを把握するための2つの方法であり、これらはスペクトルの両極端に位置しています。ポーリングはプル型であり、アプリケーションがノードに対して新しいデータが存在するかどうかを繰り返し問い合わせます。一方、ストリーミングはプッシュ型であり、新しいデータが利用可能になった瞬間にサービスから配信されます。以下の表では、アプローチを選択する際に最も重要な観点について、これら2つの方法を比較しています。

これら2つの配信モデルの詳細な比較や、それぞれが適している場面については、「ポーリングとストリーミング」をご覧ください。

ブロックチェーンによるデータストリーミングの一般的な活用事例にはどのようなものがありますか？

ストリーミングは、オンチェーンのアクティビティに迅速に対応する必要がある、あるいはデータベースを常に同期状態に保つ必要があるあらゆるワークロードに適しています。代表的な例としては、リアルタイムのウォレットやポートフォリオの追跡、DeFiダッシュボード、取引および清算ボット、NFTのミントや販売のモニタリング、不正やコンプライアンスに関するアラート、すべてのブロックを取り込むアナリティクス・ウェアハウスなどが挙げられます。これらはいずれも、定期的なスナップショットではなく、低遅延かつ途切れることのないデータ配信の恩恵を受けています。パターンのより詳細な一覧については、「ブロックチェーン・ストリーミングのユースケース」をご覧ください。

ストリーミングでは、リアルタイムデータと履歴データはどのように扱われるのでしょうか？

優れたストリーミングパイプラインは、リアルタイムデータと履歴データを、2つの別々のシステムとしてではなく、1つの連続したタイムラインとして扱います。 過去の遠い時点のブロックからストリームを開始し、履歴を遡ってバックフィルを行い、チェーンの先端に追いついた時点でライブブロックへとシームレスに移行することができます。これにより、2つの異なるコードパスを持つ2つのパイプラインを維持する必要がなくなります。最新データとアーカイブデータのトレードオフについては、「リアルタイムのブロックチェーンデータと履歴データ」を参照し、バックフィル戦略については「履歴ブロックチェーンデータへのアクセス方法」を参照してください。

ストリーミングはチェーンの再編成にどのように対処するのでしょうか？

再編成は、ブロックチェーンのライブデータを処理する上で最も難しい部分です。なぜなら、アプリケーションがすでに処理したブロックが、競合するブロックに置き換えられる可能性があるからです。本番環境のストリーミングサービスでは、データをファイナリティ順に配信することでこの問題に対処しています。また、再編成が発生した際には、どのブロックがもはや正規のものではなくなったかを宛先に通知する修正ペイロードを送信し、影響を受けたレコードをロールバックできるようにしています。これにより、カスタムパイプラインにおいて最もエラーが発生しやすい要素の1つが排除されます。詳細については、「ブロックチェーンの再編成とは」をご覧ください。

ストリーミングとインデックス作成の違いは何ですか？

ストリーミングとインデックス作成は、関連してはいるものの、それぞれ異なる問題を解決します。ストリーミングは配信に関するもので、オンチェーンデータを発生と同時に宛先へ届けることを指します。一方、インデックス作成は整理に関するもので、そのデータをクエリ可能なデータベースとして構造化し、過去のデータや集計データのクエリを高速に実行できるようにするものです。多くのチームでは、ストリーミングを利用してインデックスにデータを格納し、そのインデックスに対してクエリを実行して分析を行っています。これらの要素がどのように連携しているかについては、「ブロックチェーンのインデックス作成とは」および「ブロックチェーンデータのクエリ」をご覧ください。

よくある質問

ブロックチェーンのデータストリーミングは、WebSocketのサブスクリプションと同じものですか？

そうとは限りません。WebSocketのサブスクリプションは、ソケットが開いている間、イベントをプッシュする低レベルの接続ですが、配信の保証はなく、接続が切断されると停止してしまいます。本番環境向けのストリーミングでは、この基本的なプッシュモデルに加え、正確に1回限りの配信、ファイナリティ順序、再編成による修正、サーバーサイドのフィルタリング、および履歴のバックフィル機能が提供されます。

ストリーミングを利用すれば、ブロックを見逃すことはないのでしょうか？

はい、本番環境向けのサービスであれば可能です。「配信保証」とは、各ブロックのデータが宛先で確実に受信されるまで、プロバイダー側がバッファリングを行い、再送信を繰り返すことを意味します。そのため、エンドポイントが一時的に利用不能になった場合でも、途切れることのないフィードが提供されます。Quicknode Webhooksのようなプッシュ型イベントサービスも、イベント通知において同様の信頼性原則を採用しています。

ストリーミングデータが宛先に到達する前に、フィルタリングすることはできますか？

はい。サーバーサイドのフィルタリングでは、アドレス、イベントシグネチャ、関数セレクタ、値、あるいはブロックのその他のプロパティに基づいて照合を行うことができるため、必要なレコードのみが配信されます。これにより、不要なデータを受け取ることがないため、帯域幅、ストレージ、および処理コストを削減できます。

ストリーミングは、カスタムETLパイプラインに取って代わることができるでしょうか？

ほとんどのチームにとっては、その通りです。マネージド・ストリーミング・サービスは、データの抽出、フィルタリング、変換、配信、再試行、順序付け、再編成の処理を1つのパイプラインに統合し、自社開発のETLスタックに存在する多くの複雑な要素を排除します。ユーザーは、どのデータをどこに送信するかを設定するだけで、残りの処理はプロバイダーが担当します。

ストリーミングデータは、どの宛先に配信できますか？

一般的な送信先としては、Webhook URL、PostgreSQL、Snowflake、Amazon S3、Azure Blob Storageなどが挙げられます。これにより、それぞれにカスタムコネクタを構築することなく、ストリーミングデータをアプリケーションのバックエンド、データウェアハウス、またはオブジェクトストレージに直接取り込むことができます。

参考資料

Streams の始め方- Quicknode ドキュメント
コスト削減とブロックチェーンデータパイプラインの簡素化- Quicknode ブログ
ストリーミングインデックスデータで製品開発を加速- Quicknode ブログ
Quicknode Streams