Cloudflare WorkersDeveloper WeekR2開発者プラットフォーム

新着：ArroyoとPipelinesを使用したCloudflareへのストリーミング取り込み

Micah Wylde、Matt Silverlock、Pranshu Maheshwari

14分で読了

この記事は以下でも利用可能です English、简体中文.

BLOG-2785 Feature Image

本日、当社のストリーミング取り込み製品であるPipelinesのオープンベータ版の提供を開始します。パイプラインを使用すると、大量の構造化されたリアルタイムデータを取り込み、オブジェクトストレージサービスである R2 にロードすることができます。基盤となるインフラストラクチャを管理する必要はなく、シャードやメタデータサービスのスケーリングについて心配する必要もありません。また、処理されたデータに対して料金を支払います（時間単位ではありません）。Workers有料プランをご利用であれば、どなたでもご利用いただけます。1秒あたり数万リクエスト（RPS）のデータをR2に直接取り込み、バッチ処理することができます。

しかし、これは氷山の一角に過ぎません。多くの場合、取り込んでいるデータを変換し、他のソースからそのデータをリアルタイムで抽出し、オープンテーブルフォーマット（Apacheicebergなど）に書き込んで、オブジェクトストレージに格納した データを、効率的にクエリーを実行できます。

良いニュースは、当社でもこの点について考えており、それを実現するために、クラウドネイティブの分散型ストリーム処理エンジンである Arroyo を買収したことを発表できることを嬉しく思います。

Arroyoおよび最近発表されたR2 Data Catalogを使用して、地球全体からデータを取り込み、大規模に保存し、計算処理を実行できるようなデータプラットフォームの構築にますます真剣に取り組んでいます。

まずは、Pipelinesの開発者向けドキュメントをご覧いただくか、このWranglerコマンドを実行して最初のパイプラインを作成してください。

$ npx wrangler@latest pipelines create my-clickstream-pipeline --r2-bucket my-bucket

...
✅ Successfully created Pipeline my-clickstream-pipeline with ID 0e00c5ff09b34d018152af98d06f5a1xv

...そして、最初のレコードを書きます：

$ curl -d '[{"payload": [],"id":"abc-def"}]' 
"https://0e00c5ff09b34d018152af98d06f5a1xvc.pipelines.cloudflarestorage.com/"

しかし、真のパワーは、取り込みとR2のようなシンクに書き込まれるときの間のデータストリームの処理から来ます。取り込み ているデータのウィンドウに作用するSQLを書き、変換・集約し、さらにはデータからリアルタイムでインサイトを抽出することもできるため、非常に強力であることが判明します。

ここでArroyoが登場します。Arroyoの優れた部分をPipelinesに取り込み、 Workers、R2、および他の開発者プラットフォームと深く統合することになります。

Arroyoの原点

（Arroyo創設者 Micah Wyldeより）

当社は、データで働くすべての人にリアルタイム（ストリーム）処理を提供するために、2023年にArroyoを立ち上げました。現代企業は、ユーザーカスタマイズ、レコメンデーション、不正対策、新しいAIエージェントの世界など、アプリケーションとビジネスを強化するために、データパイプラインに依存しています。

しかし現在では、これらのパイプラインのほとんどはバッチ処理で稼働しており、1時間に1回、1日に1回、あるいは1か月に1回稼働しています。その理由は、LyftやSplunkのような企業で長年ストリーム処理に取り組んできた結果、疑問の余地がありませんでした。開発者やデータサイエンティストにとって、正しく、機能性、信頼性の高いパイプラインを構築することは、あまりにも困難でした。大規模テクノロジー企業はストリーミングの専門家を雇ってこれらのシステムを構築・運用していますが、それ以外の企業はバッチが到着するのを待つしかありません。

設立当時、ストリーミングパイプラインの主流的なソリューションはApache Flinkであり、LyftやSplunkで実行していました。Flinkは、フォールトトラレント（障害から一貫して回復できる）で、分散型（複数のマシンにまたがる）でステートフル（過去のイベントに関するデータを記憶する）なデータフローと グラフ構築APIをうまく組み合わせる最初のシステムでした。このような機能の組み合わせにより、ウィンドウ、集計、結合などの機能を備えた強力なリアルタイムデータアプリケーションを構築できるようになりました。しかし、Flinkには必要なパワーがあったものの、実際にはAPIは専門家でないユーザーにとってはあまりにも難しく、低レベルであることが判明し、結果として得られるサービスのステートフルな性質は無限の操作を必要としました。

当社は、新しいストリーミングエンジンを構築する必要があることに気づきました。Flinkのパワーを備え、製品エンジニアやデータサイエンティスト向けに設計され、最新のクラウドストラクチャで実行できるエンジンです。使いやすく、広く知られており、宣言的であるため、 APIとしてSQLを使用することにしました。CloudflareはRustで構築し、スピードと運用のシンプルさを実現しました（JVMのチューニングは不要です！）。Cloudflareではオブジェクトストレージネイティブのステートバックエンドを構築し、ステートフルパイプラインの実行という課題を簡素化しました。パイプラインはそれぞれが奇妙で専用のデータベースのようなものです。そして、2023年の夏、オープンソース化しました。現在、数十社がArroyoパイプラインを稼働し、データの取り込み、不正行為防止、 IoTの可観測性、金融取引などのユースケースを用いています。

しかし、エンジンはパズルの1ピースであることは常に認識していました。バッチのように簡単にストリーミングを行うには、ユーザーはクラスタのサイズや継続的な操作を気にすることなく、クエリロジックの開発とテスト、過去のデータでのバックフィル、サーバーレスデプロイを行う必要があります。ストリーミングの民主化は、完全なデータプラットフォームを構築する必要がありました。そして、 Cloudflareに問い合わせてみたところ、同社はすでにすべてのピースを持ち込んでいることがわかりました。R2は状態と保管中のデータを保存するためのオブジェクトストレージ、転送中のデータのためのCloudflare Queues、そして安全かつ効率的にユーザーコードを実行するためのWorkersを提供します。そして、 Cloudflareは、独自のものであり、これらのシステムをエッジまで推進し、 データ主権とAI の未来にとって鍵となるローカルストリーム処理の新しいパラダイムを可能にします。

だからこそ、 Cloudflareチームと共に、このビジョンを実現することを大変嬉しく思います。

大規模な取り込み

Pipelinesは、変換とSQLストリーミングAPIの実装が予定されていますが、データジャーニーの2つの重要な部分、グローバルに分散、高スループットの取り込み、およびオブジェクトストレージへの効率的な読み込みはすでに解決しています。

パイプラインの作成は、1つのコマンドを実行するのと同じくらい簡単です：

$ npx wrangler@latest pipelines create my-clickstream-pipeline --r2-bucket my-bucket

🌀 Creating pipeline named "my-clickstream-pipeline"
✅ Successfully created pipeline my-clickstream-pipeline with ID 
0e00c5ff09b34d018152af98d06f5a1xvc

Id:    0e00c5ff09b34d018152af98d06f5a1xvc
Name:  my-clickstream-pipeline
Sources:
  HTTP:
    Endpoint:        https://0e00c5ff09b34d018152af98d06f5a1xvc.pipelines.cloudflare.com/
    Authentication:  off
    Format:          JSON
  Worker:
    Format:  JSON
Destination:
  Type:         R2
  Bucket:       my-bucket
  Format:       newline-delimited JSON
  Compression:  GZIP
Batch hints:
  Max bytes:     100 MB
  Max duration:  300 seconds
  Max records:   100,000

🎉 You can now send data to your pipeline!

Send data to your pipeline's HTTP endpoint:
curl "https://0e00c5ff09b34d018152af98d06f5a1xvc.pipelines.cloudflare.com/" -d '[{ ...JSON_DATA... }]'

デフォルトでは、パイプラインはWorkersとHTTPエンドポイントの2つのソースからデータを取り込み、バッチ処理されたイベントを R2バケットにロードできます。これにより、未加工のイベントデータをオブジェクトストレージにストリーミングするための、すぐに使えるソリューションが提供されます。デフォルトが機能しない場合は、作成中、または作成後いつでもパイプラインを設定できます。HTTPエンドポイントへの認証の追加、ブラウザがオリジン間リクエストを行えるようにCORSの設定、出力ファイルの圧縮とバッチ設定の指定などがあります。

初日から大量の取り込みのためのパイプラインを構築しました。各パイプラインは毎秒100,000レコードまで拡張できます（ここではまだ始まったばかりです）。レコードがパイプラインに書き込まれると、耐久性のある格納、バッチ処理、および R2バケット内のファイルとしての書き込みが行われます。ここではバッチ処理が重要になります。もしデータに基づいてクエリーを実行するのであれば、クエリーエンジンに何百万もの（あるいは何千万もの）小さなファイルをクエリーさせる必要はありません。低速（ファイル単位およびリクエストのオーバーヘッド）、非効率性（読み込むファイルの数が増える）、コストがかかる（より多くの操作）。その代わり、クエリエンジンのバッチサイズと遅延の適正なバランスを見つける必要があります。パイプラインでは、これを設定することができます。

クエリをさらに最適化するために、出力ファイルは標準的なHive分割スキームを使用して、日付と時間で分割されます。クエリーエンジンは、実行中のクエリーと無関係なデータをスキップすることができるため、クエリーをさらに最適化することができます。R2バケットへの出力は以下のようになります。

BLOG-2785 Image 1

R2バケットのパイプラインからハイ分割されたファイル

出力ファイルは、新しい行で区切られたJSON（NDJSON）として保存され、これらのファイルからストリームを具体化することが容易になります（ヒント：将来的にはR2をパイプラインソースとして使用できるようになります）。最後に、ファイル名はULIDであるため、デフォルトで時間でソートされています。

最初にシャードネットワークに接続し、

Pipelinesを水平方向にスケーラブルにし、書き込みを素早く認識できるようにしているのは、当社の構築方法です。Durable Durable Objectsと各Durable Object内の埋め込まれたゼロ遅延SQLiteストレージを使用して、書き込まれたデータをすぐに保持し、その後、処理して書き込む前に、 R2にします。

たとえば、eコマースまたはSaaSサイトであり、 Webサイトの利用状況データ（クリックストリームデータと呼ばれる）を取り込み、データサイエンスチームがクエリできるようにする必要があるとします。このワークロードを処理するインフラストラクチャは、いくつかの障害シナリオに耐性がある必要があります。取り込みサービスは、トラフィックの急増に直面しても高い可用性を維持する必要があります。取り込んだデータはバッファリングして、ダウンストリームの呼び出しを最小限に抑え、ダウンストリームのコストを最小限に抑える必要があります。最後に、バッファリングされたデータは、シンクが利用できない場合、適切なリトライおよび失敗処理を行って、シンクに提供する必要があります。このプロセスの各ステップは、過負荷時にバックプレッシャーをアップストリームに知らせる必要があります。また、スケールも必要で、大規模なセールやイベント中は増加し、1日の中で静かな時間帯には減少しています。

この記事を読んでいるデータエンジニアの皆さんは、Kafkaとその処理に関連するエコシステムの使用の現状をご存知かもしれません。しかし、あなたがアプリケーションエンジニアであれば、Kafka、Zookeeper、Kafkaストリームについて学ぶことなく、Pipelinesを使用して取り込みサービスを構築することになります。

BLOG-2785 Image 2

パイプラインの水平シャーディング

上の図は、Pipelinesが、課金、追跡シャード、パイプラインのライフサイクルイベントを担当するコントロールプレーンと、 Durable Objectsシャードのスケーラブルなグループであるデータパスを分割する方法を示しています。

レコード（またはレコードのバッチ）がパイプラインに書き込まれると：

  1. Pipelines Workerは、フェッチハンドラーまたはworkerバインディングを通じてレコードを受け取ります。
  2. Pipeline_idに基づいて、コーディネーターに連絡を取り、実行プランを取得します。それ以降の読み取りは、コーディネーターの負荷を軽減するためにキャッシュされます。
  3. 主に読み取りリクエストの処理をスケール化し、
  4. これらはその後、実際に書き込みを処理する別のエグゼクティブにリシャードされます。Durable Object ストレージに保持されることから始め、これはストレージリレーサービス（SRS）によって複製され、耐久性と可用性が確保されます。
  5. SRSの後、設定されたTransform Workersに移り、データをカスタマイズします。
  6. データはバッチ処理され、出力ファイルに書き込まれ、（該当する場合）圧縮されます。
  7. ファイルは圧縮され、データは最終的なバッチにパッケージ化され、設定された R2バケットに書き込まれます。

このパイプラインの各ステップは、アップストリームにバックプレッシャーを知らせることができます。これは、ReadableStreamsを活用し、書き込み待ちのバイト数の総数がしきい値を超えた場合に429で応答することによって行われます。各ReadableStreamは、 Durable Objects間でJSRPCコールを使用することで、 Durable Object の境界を越えることができます。パフォーマンスを向上させるために、 Durable Objects間で接続を再利用するためにRPCスタブを使用します。各ステップは、Durable ObjectsまたはR2が一時的に使用できない場合に対処するために、操作を再試行することもできます。

また、既存のパイプラインを更新している間でも配信を保証します。既存のパイプラインを更新すると、上記のすべてのシャードとDurable Objectsを含む新しいデプロイメントが作成されます。リクエストは新しいパイプラインに安全に再ルーティングされます。古いパイプラインは、すべてのDurable Objectストレージが枯渇するまで、R2にデータを書き込み続けます。すべてのデータが書き込まれた後にのみ、古いパイプラインをスピンダウンします。こうすることで、パイプラインを更新しながらデータを失うことはありません。

ここに、まだ公開していない興味深い部分があることにお気づきいただけるかと思います。ArroyoのストリーミングエンジンをPipelinesと統合することに取り組んでおり、これはArroyoへのデータの引き渡し方法において重要な部分となります。

では、そのコストは？

オープンベータの第1段階では、データの読み込み時やアクセス時に発生する標準R2のストレージ料金と運用コスト以外の追加料金は発生しません。また、R2バケットからの直接のエグレスは無料ですので、データ転送コストの追加を心配することなく、どのクラウドや地域からでもデータを処理し、クエリーを行うことができます。

将来的には、Pipelineに取り込まれ、Pipelineから配信されるデータの量に基づく料金設定を導入する予定です。


Workers有料版 ($5 /月)

取り込み

月々50GBまで込み追加GBごとに$0.02

R2への配信

月々50GBまで込み追加GBごとに$0.02

また、ベータ版の進捗に合わせて、PipelinesをWorkers Freeプランで利用できるようにする予定です。

Pipelinesに変換や追加のシンクを導入する際に、より多くのことを共有していきます。変更を行う前、または使用量に対する課金を開始する前に、少なくとも30日前に通知を行います。この通知は、2025年9月15日までに行う予定です。

今後の展開は？

ここには多くのことを構築する必要がありますし、Arroyoが構築した多くの強力なコンポーネントを基に構築したいと考えています。UDF（ユーザー定義機能）としてのWorkersの統合、Kafkaクライアントのような新しいソースの追加、新しいシンクでのパイプラインの拡張などです。 （R2以上）。

また、Pipelinesを ローンチしたばかりの R2 Data Catalog と統合します。Icebergテーブルに直接データストリームを取り込み、すぐにクエリーを実行できるようにします。他のシステムに依存する必要はありません。

同時に、以下が可能です：

...または、サンプルプロジェクトを直接デプロイする：

$ npm create cloudflare@latest -- pipelines-starter 
--template="cloudflare/pipelines-starter"