Cloudflare で始める Data Platform

「みんなの考えた最強のデータ基盤アーキテクチャ'26前期〜前夜祭〜ルーキーズ※公式」で登壇しました。
https://datatech-jp.connpass.com/event/386885/

生の資料↓
https://slides.ta93abe.com/cloudflare-data-platform/

Transcript

1. Cloudflare で始める Data Platform 阿部拓海
2. Cloudflare とは CDN？エッジコンピューティング？
3. Cloudflare Data Platform Cloudflare の Cloudflare Data Platform は、入れる/貯める/使うを 1 つのプラットフォームで提供します。 (Announcing the Cloudflare Data Platform: ingest, store, and query your data directly on Cloudflare)
4. Pipelines 2025年4月に買収した Arroyo をベースとしたストリーミングインジェストサービスです。 Streams で HTTP / Workers Binding / Logpush からデータを受けます。 Pipelines で SQL 変換を行えます。 Sinks で --roll-size or --roll-interval で設定した粒度で自動バッチ化し、R2 / R2 Data Catalog に書 き出せます。 wrangler pipelines setup
5. R2 wrangler r2 bucket create < bucket-name >
6. R2 Data Catalog データを 構造化する レイヤーです。R2 上の Apache Iceberg テーブルをマネージドで管理します。 Trino / DuckDB / PyIceberg / Snowflake / Spark / StarRocks などのクライアントから直接クエリ可能 ACID / Schema evolution / Time travel などの Iceberg らしい機能はもちろん対応している。 テーブルメンテナンス Compaction: --target-size で指定したサイズ に合わせて Parquet ファイルを集約 Snapshot expiration: --older-than-days で古 いスナップショットを削除、 --retain-last で 最低限残す数を指定 wrangler r2 bucket catalog enable < bucket-name >
7. R2 SQL R2 Data Catalog の Iceberg テーブルに標準 SQL を実行できる、Cloudflare ネイティブの分散クエリエンジン です。Apache DataFusion をベースにしています。 基本的な演算はできますが、JOIN や WINDOW 関数はまだ対応していません。ベータ版で開発真っ只中。 実行方法は Wrangler と HTTP API の 2 つがあります。管理画面などに Web SQL エディターはありません。 wrangler r2 sql query "$WAREHOUSE" \ "SELECT user_id, COUNT(*) AS n FROM default.events WHERE __ingest_ts > '2026-05-01' GROUP BY user_id LIMIT 10" curl -X POST \ "https://api.sql.cloudflarestorage.com/api/v1/accounts/{ACCOUNT_ID}/r2-sql/query/{BUCKET_NAME}" \ -H "Authorization: Bearer {API_TOKEN}" \ -H "Content-Type: application/json" \ -d {"query": "SELECT user_id, COUNT(*) AS n FROM default.events WHERE __ingest_ts > '2026-05-01' GROUP BY user_id LIMIT 10"}
8. Cloudflare Workflows Cloudflare Workflows は耐久性のある実行エンジンです。ステップを連鎖させ、失敗時には自動で再試行し、 長期間実行されるプロセス全体で状態を保持します。各 step には Workers Bindings を組み込めます。 1. R2 から画像を取得 2. LLaVA で 1 文の説明を生成 3. `approved` イベントを最大 24h 待機 4. R2 へ publish (公開ディレクトリ) export class ImageProcessingWorkflow extends WorkflowEntrypoint { async run(event: WorkflowEvent, step: WorkflowStep) { const imageData = await step.do('fetch image', async () => { const object = await this.env.BUCKET.get(event.params.imageKey); return await object.arrayBuffer(); }); const description = await step.do('generate description', async () => { const imageArray = Array.from(new Uint8Array(imageData)); return await this.env.AI.run('@cf/llava-hf/llava-1.5-7b-hf', { image: imageArray, prompt: 'Describe this image in one sentence', max_tokens: 50, }); }); await step.waitForEvent('await approval', { type: 'approved', timeout: '24 hours', }); await step.do('publish', async () => { await this.env.BUCKET.put(`public/${event.params.imageKey}`, imageData); }); } }
9. Python SDK WorkflowEntrypoint を Python で継承します。関数パラメータ名で依存を暗黙解決 する DAG 表現が特徴で す。引数名による暗黙的依存解決で DAG が宣言的に書けます。 1. `fetch_a` を `@step.do()` で定義 2. `fetch_b` を独立した step として定義 3. `merge` を `concurrent=True` + 引数名 `fetch_a` / `fetch_b` で依存宣言 4. `await merge()` 実行 — 依存先が並列起動 from workers import WorkflowEntrypoint class IngestWorkflow(WorkflowEntrypoint): async def run(self, event, step): @step.do() async def fetch_a(): return await get_a() @step.do() async def fetch_b(): return await get_b() @step.do(concurrent=True) async def merge(fetch_a, fetch_b): # 引数名で依存 return combine(fetch_a, fetch_b) await merge()
10. ビジュアライザ Cloudflare ダッシュボードが Workflow コードを parse し、step / 並列 / 条件分岐 / ループのフロー図 を自動 生成します。
11. Cloudflare Containers Cloudflare Workers では 128 MB の実行メモリ制限や CPU時間制限があります。 そこで Cloudflare Containers を使えば、この制約を突 破できます。 例えば dbt の実行を行えます。バッチデ ータインジェストがしたいなら dlt を使うと良いでし ょう。 Cloudflare で完結させるメリットは次のとおりです。 dbt artifacts を R2 に Binding 経由で永続化 ( env.BUCKET.put('*.json', body) ) Workers が R2 Binding 経由で dbt docs を配信 Cloudflare Access で社内限定配信 Workers Observability でログを一元管理
12. まとめ まずは cloudflare.com にたどり着きましょう。 無事たどり着いて管理画面に入ったら Agent Lee が迎えてくれます。 こんな Cloudflare の始め方もあります。 ドメインを購入する or 移管してくる。 個人のサイト/ブログを Workers Static Assets でホスティングしてみる。 R2 Data Catalog は Iceberg を始めるには結構お手軽です。 AI エージェントを実装してみる。 Agent / Workers AI / AI Gateway / Dynamic Workers / Sandbox / AI Search / Browser Run / Artifacts / Durable Objects