Cloudflare開発ツールインフラ

Cloudflare、K2を発表：サーバーレスイベントストリーミング

K2はプロデューサーとコンシューマーをスケールと時間の両面で分離します。エッジでの永続バッファリング、複数コンシューマーによる独立した読み取り、キー単位の順序保証、Kafkaクライアントとの互換性を備えます。本記事ではQueuesやPipelinesとの役割分担と適用範囲を解説します。

Cloudflare Blog約 9 分

日本語
中文日本語
コピー
Cloudflare 风格的插画：橙色放大镜里是一座雪山，四周有数据流线条、彩色方块与节点

Cloudflare、K2 を発表：サーバーレスなイベントストリーミング

従来のリモートプロシージャコール（RPC）アーキテクチャには、核心となる難題がひとつある。プロデューサーとコンシューマーが規模とペースを揃えなければならない点だ。プロデューサーがコンシューマーの処理能力を超えるデータを送れば、あるいはコンシューマーやその下流のサービスが利用できなければ、イベントは失われる。複数のコンシューマーがそれぞれ独立してデータを処理する必要がある場合は、さらに厄介になる。たとえば EC のバックエンドは取引の完了時にイベントを発行し、それを分析システムと不正検知サービスが読む必要がある。解決策はプロデューサーとコンシューマーを分離すること、つまり間にサービスを挟んで書き込みを引き受けさせ、各読み手が自分のペースで独立して消費できるようにすることだ。この問題を解くために、本日 Cloudflare K2 のオープンベータ版を公開する。K2 は Developer Platform 上の永続的なイベントストリーミングプリミティブだ。イベントを K2 ストリームに送ると、順序付きログとして保存される。コンシューマー側の読み取り方は複数ある。読み取りタスクをコンシューマーのグループに分割したり、すべてのメッセージを全コンシューマーに配信したりできる。完全にサーバーレスで、大量データまでスケールし、長期保持にも対応するため、コンシューマーが長期間止まってもデータは失われない。内部では、K2 が R2 オブジェクトストレージ上にパーティション化された永続ログを実装しており、これが巨大なストレージ規模への拡張を可能にしている。使い始める準備ができたら、こちらのガイドに従えば数秒で最初のストリームを作成できる。

エッジ上のストリーム

K2 を作り始めたのは、エッジ上に永続バッファが必要だったからだ。当初は Basin Pipelines の取り込み層として始まった。Pipelines はストリーム処理エンジンによって駆動され、プルモデルを採用している。つまり、イベントを読み取り、変換し、R2 に書き込む前に、別のシステムが先にイベントを保存しておく必要がある。一度 Pipelines Stream が受け取ったイベントは決して捨てないと約束している以上、そのストレージは永続でなければならない。データを失わず、場合によっては非常に長い時間を耐えられる必要がある。多くの企業はこの段階で Apache Kafka をデプロイする。だが Pipelines は Cloudflare のエッジ上、335 以上の都市にまたがる大量のサーバーで動いている。この独特なアーキテクチャでは Kafka のような従来の分散システムソフトウェアは動かせないことが多く、こうしたシステムの構築と運用を根本から考え直す必要がある。ステートフルなサービスでは特にそうだ。Cloudflare のグローバルインフラはいくつかの課題を持ち込む。手に入るマシンのスライスは比較的小さく、マシン自体の寿命は短く、ネットワークはしばしば公衆網を通る。一方で、このインフラには強力な利点もある。ユーザーがどこにいても近くにいられること、そして驚異的な水平スケール能力だ。後に K2 となる永続バッファを設計するにあたり、私たちは既にある強力なステートプリミティブ、R2 に頼ることにした。R2 のようなオブジェクトストレージは、極めて高い耐久性（イレブンナイン！）と強整合な API を兼ね備えている。レプリケーションとコンセンサスをストレージ層に押し下げれば、アプリケーション層（ここでは K2）はずっと単純に、安く、高性能になる。副次的な利点として計算とストレージが分離され、それぞれを独立にスケールできる。これにより大量の履歴データを低コストで保存できる。オブジェクトストレージの上でログをどう作るか。すぐにぶつかる問題は、R2 が他のオブジェクトストレージと同様に append をサポートしないことだ。append はログの標準操作である。書けるのは完全なファイル、つまり segment だけで、しかもファイルごとの読み書きのオーバーヘッドを薄めるために十分大きくする必要がある。そこで、まずエッジサービスのメモリ上に書き込みを蓄積する。少し待ってデータを揃えてから、すべてのイベントをひとつの segment ファイルとして書き出す。順序と厳密に増加する offset は R2 のアトミック操作を使って実現し、追加のコーディネーションサービスを必要としない。R2 の上に構築することには多くの利点があるが、欠点がひとつある。書き込みレイテンシが高くなるのだ。オブジェクトストレージへの書き込みはローカルディスクより遅く、しかもローカルのバッチが溜まるまで書き始められない。K2 の最初のバージョンでは、これが合わさって 99 パーセンタイルで約 1 秒の書き込みレイテンシになった。K2 の設計の詳細については、今後の技術深掘り記事でさらに共有する。

Streams、Queues、それとも Pipelines？

Cloudflare にはすでに非同期配信のプリミティブがいくつかあり、Queues や Basin Pipelines もそのひとつだ。既存のこれらの製品ではなく K2 をいつ使うべきか。Queues と K2 Streams は表面的には似ている。どちらもイベントを受け取り、永続的に保存し、コンシューマーに配信する。Queues の設計は個々のタスク項目の追跡を中心にしている。非同期で完了させる必要のある、コストの高い、あるいは時間のかかる処理だ。たとえば画像処理アプリはユーザーのリクエストをキューに入れ、実際の画像処理サービスに処理させる。個々の作業項目に対する複雑なロジック、たとえばリトライ、遅延、失敗時のデッドレターキューをサポートする。K2 は違う。大規模なデータ移動、長期保持、ファンアウト消費を狙っている。メッセージはバッチ単位で生成・消費される。効率的な処理と引き換えに、メッセージ単位のリトライはできない。このバッチ処理のため、プロデューサーのレイテンシも Queues より高い。Basin Pipelines はサーバーレスの取り込みサービスだ。Pipeline に JSON イベントを送ると、変換を経て R2 または Basin Catalog に書き込まれる。最終的な目的がイベントをオブジェクトストレージや Iceberg テーブルに書き込むことなら Pipelines を勧める。独自の処理をしたい場合や別の宛先に書きたい場合は K2 を使えばいい。

はじめに

K2 を使うには、まず stream を作成する必要があります。1 つのアカウントに複数の stream を作成でき、それぞれを異なるユースケースやイベントタイプに対応させられます。作成方法は cf、Wrangler、ダッシュボード、API のいずれかです。ここではプロダクト分析データの収集と処理を例にします。まず cf で stream を作成します。

$ cf k2 streams create --name app_events --http-enabled

{
  "id": "d78b09ee1f50430e9ec92a8af92b0231",
  "name": "app_events",
  "retention_seconds": 604800,
  "endpoint": "https://d78b09ee1f50430e9ec92a8af92b0231.k2.cloudflarestorage.com",
  "http": {
    "enabled": true,
    "authentication": false
  },
  "worker_binding": {
    "enabled": true
  },
  "created_at": "2026-09-28T15:14:39.053Z",
  "modified_at": "2026-09-28T15:14:39.053Z"
}

stream ができたら、あとはデータを書き込むだけです。HTTP API でも Worker binding でもかまいません。たとえば Worker の中では次のように書けます。

const result = await env.EVENTS.send([
  {
    content: new TextEncoder().encode(
      JSON.stringify({
        event: "page_view",
        path: new URL(request.url).pathname,
        timestamp: Date.now(),
      }),
    ),
    headers: { "content-type": "application/json" },
  },
]);

if (!result.success) {
  console.error(`Produce failed: ${result.error.message}`);
  return new Response("Failed to record event", {
    status: result.error.retryable ? 503 : 500,
  });
}

K2 はデータをバイト列として扱うため、フォーマットとエンコーディングはアプリケーションの要件に合わせて自由に決められます。

stream にイベントが入ったら、次は subscription を作成します。Subscription は処理を複数の consumer に分配し、読み取りの並列性 を実現します。つまり読み取り側を水平にスケールさせ、1 台のサーバーでは捌けない負荷に耐えられるようにする仕組みです。

subscription は HTTP API から作成できます。

$ curl -X POST "https://d78b09ee1f50430e9ec92a8af92b0231.k2.cloudflarestorage.com/subscriptions" \
    -H "Authorization: Bearer ${CLOUDFLARE_API_TOKEN}" \
    -H "Content-Type: application/json" \
    --data '{
      "name": "analytics_processor",
      "start_at": { "type": "earliest" }
    }'
{
  "result": { "id": "ee13f761783d3823a447a47b572ebf76" },
  "success": true,
  "errors": [],
  "messages": []
}

作成した subscription は、各 consumer からポーリングします。

$ curl -X POST \"https://4d8f5394e3e733debdeca9c65c5b7439.k2.cloudflarestorage.com/subscriptions/ee13f761783d3823a447a47b572ebf76/consume" \
    -H "Authorization: Bearer ${CLOUDFLARE_API_TOKEN}" \
    -H "Content-Type: application/json" \
    --data '{
      "worker_id": "analytics-1",
      "max_records": 100
    }' 
{
  "result": {
    "batch_id": "b7e4c9210a3f468d95c2e1068fdb734a",
    "leased_until_ms": 1790633929437,
    "records": [
      {
        "timestamp_ms": 1790633629168,
        "content": "eyJldmVudCI6InBhZ2VfdmlldyJ9",
        "headers": {
          "content-type": "application/json"
        }
      },
      ...
    ]
  },
  "success": true,
  "errors": [],
  "messages": []
}

クライアントが consume を呼び出すと、そのバッチのイベントに対する lease を取得します。有効期間は 5 分です。クライアントは次の 3 つのいずれかを行えます。

  • ack — バッチを処理済みとしてマークし、再配信されないようにする
  • nack（negative ack）— 処理に失敗したことを示し、再配信を求める
  • extend — lease を延長する。処理時間が足りない場合に使う
$ curl -X POST "https://d78b09ee1f50430e9ec92a8af92b0231.k2.cloudflarestorage.com/subscriptions/ee13f761783d3823a447a47b572ebf76/batches/b7e4c9210a3f468d95c2e1068fdb734a/ack" \
  -H "Authorization: Bearer ${CLOUDFLARE_API_TOKEN}" \
  -H "Content-Type: application/json" \
  --data '{ "worker_id": "analytics-1" }'

これが K2 での消費方法の 1 つで、処理を複数の consumer に分割し、各 consumer はデータの一部だけを受け取ります。もう 1 つは consumer ごとに別々の subscription を使う方法、つまり pub/sub パターンで、こちらでは各 consumer がすべてのメッセージを受け取れます。両方を組み合わせ、互いに独立した consumer プールを複数維持することも可能です。

API の詳細は K2 のドキュメントを参照してください。

料金と提供状況

K2 は本日よりパブリックベータとなり、Workers Paid サブスクリプションを持つアカウントで利用できます。制限は以下のとおりです。

  • ストレージ使用量の上限は 10GB
  • stream あたりの書き込みは 30 MB/s

より高い上限が必要な場合は、Discord でチームに連絡するか、上限引き上げフォームに記入してください。

ベータ期間中、K2 の利用は無料です。課金開始後の予定価格は次のとおりです。

価格
データ書き込み$0.04 / GB
データ読み取り$0.04 / GB
データ保持$0.02 / GB / 月

今後の予定

K2 には今後数か月にわたる楽しみなロードマップがあります。

  • 書き込みの並列性を高め、stream あたり数 GB/s を実現
  • メッセージキーと、キーに基づく順序保証
  • push ベースの worker consumer
  • Express レイヤーによる書き込みレイテンシとエンドツーエンドレイテンシの削減
  • Apache Kafka クライアントとの直接互換

K2 で何を作るのか、楽しみにしています。Cloudflare Discord でのフィードバックもお待ちしています。

出典: Cloudflare Blog← ホームへ戻る