Pipelines

Pipelines 負責接收「串流資料（streaming data，指源源不絕、一筆筆小事件不停湧入的資料，例如點擊或感測器讀數）」，可選擇做轉換，再把它整整齊齊地寫進 R2 物件儲存。

一條 pipeline 有三個部分：stream（事件進來的地方，透過 HTTP 端點或 Worker）、可選的 SQL 轉換（用來過濾或重塑事件），以及 sink（資料落地的 R2 目的地，可存成 Iceberg 表、Parquet 或 JSON 檔）。

可靠地收下源源不絕的大量事件其實很難：你得緩衝突發尖峰、把它們批次寫成檔案，還不能掉資料或重複資料。Pipelines 把這些全包辦成一個受管服務，最後在 R2 給你乾淨的檔案。

引導式的設定指令會一次幫你建好 stream、SQL 轉換和 R2 sink。接著你把事件送到 HTTP 端點，再查詢存下來的資料。

1. 安裝 Wrangler 並登入

   Pipelines 在公開測試期間，於 Workers 付費方案下提供。

   bashcontent_copy複製

   npm install -g wrangler
   wrangler login

2. 執行引導式設定

   它會帶你一步步命名 pipeline、開啟 HTTP 端點、選一個 R2 儲存桶當 sink，並挑選一個轉換方式。

   bashcontent_copy複製

   npx wrangler pipelines setup --name ecommerce

3. 把事件送到端點

   用 POST 把一個 JSON 事件陣列送到你 stream 的接收 URL。把 {stream-id} 換成設定時拿到的 id。

   bashcontent_copy複製

   curl -X POST https://{stream-id}.ingest.cloudflare.com \
     -H "Content-Type: application/json" \
     -d '[{"user_id":"u_123","event":"purchase","amount":29.99}]'

4. 從 Worker 送出事件

   你也可以在 Worker 程式碼裡用 POST 打同一個接收 URL 來推送事件。

   jscontent_copy複製

   export default {
     async fetch(request, env) {
       const event = { user_id: "u_123", event: "page_view", path: "/home" };
       await fetch("https://{stream-id}.ingest.cloudflare.com", {
         method: "POST",
         headers: { "Content-Type": "application/json" },
         body: JSON.stringify([event]),
       });
       return new Response("event sent");
     },
   };

5. 查詢存下來的資料

   事件落地 R2 後，用你的 warehouse 名稱搭配 R2 SQL 來查詢它們。

   bashcontent_copy複製

   export WRANGLER_R2_SQL_AUTH_TOKEN=your_api_token
   
   npx wrangler r2 sql query "YOUR_WAREHOUSE_NAME" \
     "SELECT user_id, event, amount FROM default.ecommerce LIMIT 10"

• 把事件包成 JSON 陣列，就能在一個請求裡批次送出多筆紀錄。
• 若打算用大數據工具查詢，輸出格式選 Parquet 或 Iceberg。
• 因為輸出存在 R2，把資料讀回來是零流出費。
• 只是想在 Workers 之間放個簡單訊息佇列？可以看看 Queues。