Cloudflare Pipelines 配置全指南從 Worker 綁定、Streams/Sinks 到不可變 SQL 管道的完整實(shí)戰(zhàn)手冊【免費(fèi)下載鏈接】skillsSkills Catalog for Codex項(xiàng)目地址: https://gitcode.com/GitHub_Trending/skills4/skills本指南以 Cloudflare Deploy Skill 中 Pipelines 配置文檔 為骨架系統(tǒng)講解 Cloudflare PipelinesETL 流式平臺的完整配置鏈路在wrangler.jsonc中聲明 Worker 綁定、為結(jié)構(gòu)化流定義 Schema、用 CLI 創(chuàng)建 Streams 與兩類 SinkIceberg 數(shù)據(jù)目錄 / Parquet 原始存儲、創(chuàng)建不可變 SQL 管道并落地到 R2。閱讀完成后你將能夠從零搭建一條數(shù)據(jù)源 → Streams → Pipelines(SQL) → Sinks → R2的生產(chǎn)級流式 ETL 管道并掌握 Schema 校驗(yàn)、憑據(jù)配置、性能調(diào)參與常見故障排查的完整要點(diǎn)。一、Pipelines 是什么一個(gè)面向 R2 的流式 ETL 平臺Cloudflare Pipelines 是一個(gè)用于采集Ingest、轉(zhuǎn)換Transform并加載Load數(shù)據(jù)到 R2的流式 ETL 平臺。其核心架構(gòu)由三部分組成見 Pipelines READMEData Sources → Streams → Pipelines (SQL) → Sinks → R2 ↑ ↓ ↓ HTTP/Workers Transform Iceberg/Parquet組件職責(zé)關(guān)鍵特性Streams事件采集持久的緩沖層HTTP / Workers 寫入支持結(jié)構(gòu)化帶 Schema 校驗(yàn)與非結(jié)構(gòu)化兩種模式Pipelines用 SQL 對流做轉(zhuǎn)換創(chuàng)建后不可變immutable無法修改 SQLSinks將數(shù)據(jù)寫入 R2 目的地提供精確一次exactly-once投遞語義常見的落地形態(tài)分析管道點(diǎn)擊流、遙測、服務(wù)器日志、數(shù)據(jù)倉庫ETL 進(jìn)可查詢的 Iceberg 表、事件處理移動端 / IoT 富化、電商分析用戶事件、購買、瀏覽。當(dāng)前狀態(tài)為Open Beta需要 Workers Paid 套餐除標(biāo)準(zhǔn) R2 存儲/操作費(fèi)用外不額外計(jì)費(fèi)以倉庫文檔聲明為準(zhǔn)。二、Worker Binding在 wrangler.jsonc 中聲明流綁定要讓 Worker 代碼能夠?qū)懭肓魇紫刃枰趙rangler.jsonc中聲明Pipelines 綁定。配置文檔給出的完整示例// wrangler.jsonc { pipelines: [ { pipeline: STREAM_ID, binding: STREAM } ] }要點(diǎn)解析pipeline字段填的是Stream ID流 ID而不是 Pipeline管道ID——這是最容易踩的坑。通過以下命令獲取流 IDnpx wrangler pipelines streams listbinding是你自己在 Worker 代碼中使用的環(huán)境變量名如STREAM之后在Env類型中聲明并調(diào)用env.STREAM.send(...)即可寫入事件。修改綁定后必須重新部署npx wrangler deploy才會生效若出現(xiàn)env.STREAM is undefined優(yōu)先檢查這兩點(diǎn)詳見 gotchas.md。綁定完成后Worker 中的最小寫入示例來自 api.mdinterface Env { STREAM: Pipeline; } export default { async fetch(request: Request, env: Env, ctx: ExecutionContext): PromiseResponse { const event { user_id: 123, event_type: purchase, amount: 29.99 }; // Fire-and-forget 模式不阻塞響應(yīng) ctx.waitUntil(env.STREAM.send([event])); return new Response(OK); } } satisfies ExportedHandlerEnv;三、Schema為結(jié)構(gòu)化流定義字段與類型創(chuàng)建結(jié)構(gòu)化流時(shí)可以用一個(gè) JSON 文件描述事件結(jié)構(gòu)Pipelines 會據(jù)此在寫入時(shí)做字段級校驗(yàn)。配置文檔中的標(biāo)準(zhǔn) Schema 示例{ fields: [ { name: user_id, type: string, required: true }, { name: event_type, type: string, required: true }, { name: amount, type: float64, required: false }, { name: timestamp, type: timestamp, required: true } ] }每個(gè)字段由三個(gè)屬性組成屬性含義說明name字段名與 SQL 轉(zhuǎn)換中引用的列名一致type字段類型見下方支持類型清單required是否必填true缺失即校驗(yàn)失敗false允許缺省支持的類型配置文檔原文string、int32、int64、float32、float64、bool、timestamp、json、binary、list、struct?? 重要提示Gotcha結(jié)構(gòu)化流對不合法的事件會靜默丟棄——HTTP 返回 200 但事件永遠(yuǎn)不會出現(xiàn)在 Sink 中詳見 gotchas.md。因此強(qiáng)烈建議在客戶端用 Zod 先做一次校驗(yàn)獲得即時(shí)反饋import { z } from zod; const EventSchema z.object({ user_id: z.string(), event_type: z.enum([purchase, view]), amount: z.number().positive().optional() }); try { const validated EventSchema.parse(rawEvent); // 校驗(yàn)失敗會拋異常 await env.STREAM.send([validated]); } catch (e) { // 在此獲得即時(shí)錯(cuò)誤反饋 }四、Stream Setup創(chuàng)建、查詢與刪除流流的創(chuàng)建有兩種模式# 帶 Schema結(jié)構(gòu)化流寫入時(shí)校驗(yàn) npx wrangler pipelines streams create my-stream --schema-file schema.json # 不帶 Schema非結(jié)構(gòu)化流無校驗(yàn) npx wrangler pipelines streams create my-stream生命周期管理命令# 列出所有流拿到 Stream ID npx wrangler pipelines streams list # 查看單個(gè)流詳情 npx wrangler pipelines streams get ID # 刪除流注意若還有 Pipeline 引用它需先刪除管道 npx wrangler pipelines streams delete ID實(shí)戰(zhàn)提示若刪除流時(shí)提示失敗通常是因?yàn)樵摿魅员荒硞€(gè) Pipeline 引用應(yīng)先刪除管道再刪流見 gotchas.md 錯(cuò)誤對照表。向流寫入事件的兩種途徑Worker 綁定上文已述env.STREAM.send(events)支持單個(gè)對象或數(shù)組單次請求上限1 MB單流寫入速率上限5 MB/s。HTTP Ingest 端點(diǎn)適用于外部系統(tǒng)/服務(wù)端上報(bào)curl -X POST https://{stream-id}.ingest.cloudflare.com \ -H Content-Type: application/json \ -H Authorization: Bearer YOUR_API_TOKEN \ -d [{user_id: 123, event_type: purchase}]其中{stream-id}同樣來自npx wrangler pipelines streams list鑒權(quán) Token 需要Workers Pipeline Send權(quán)限D(zhuǎn)ashboard → Workers → API tokens。關(guān)鍵細(xì)節(jié)HTTP 端點(diǎn)的請求體必須是 JSON 數(shù)組而不是單個(gè)對象否則會返回 400詳見 api.md。五、Sink Configuration兩類 R2 落地目標(biāo)Sink 決定數(shù)據(jù)最終以什么形態(tài)寫入 R2。配置文檔給出了兩種類型選擇依據(jù)如下來自 README需要直接對數(shù)據(jù)跑 SQL 查詢→ 選R2 Data CatalogIceberg 表具備 ACID 事務(wù)、時(shí)間旅行time-travel、Schema 演化能力但配置更復(fù)雜需要 namespace、table、catalog token僅做文件存儲/歸檔→ 選R2 RawParquet/JSON 文件簡單直接但沒有內(nèi)置 SQL 查詢配合外部工具Spark / Athena→ 選R2 RawParquet 分區(qū)標(biāo)準(zhǔn)格式 分區(qū)裁剪提升查詢性能但 Schema 兼容性需自行維護(hù)。5.1 R2 Data CatalogIcebergSinknpx wrangler pipelines sinks create my-sink \ --type r2-data-catalog \ --bucket my-bucket --namespace default --table events \ --catalog-token $TOKEN \ --compression zstd --roll-interval 60其中--bucket指定的桶必須先啟用 Data Catalognpx wrangler r2 bucket catalog enable my-bucket詳見 r2-data-catalog 配置文檔--catalog-token為具有R2 Admin Read Write權(quán)限的 API Token。5.2 R2 RawParquetSinknpx wrangler pipelines sinks create my-sink \ --type r2 --bucket my-bucket --format parquet \ --path analytics/events \ --partitioning year%Y/month%m/day%d \ --access-key-id $KEY --secret-access-key $SECRET--partitioning使用 strftime 風(fēng)格的占位符做 Hive 風(fēng)格分區(qū)目錄方便外部查詢引擎做分區(qū)裁剪。5.3 關(guān)鍵參數(shù)速查表配置文檔原文OptionValuesGuidance--compressionzstd、snappy、gzipzstd壓縮比最佳snappy速度最快--roll-interval秒低延遲場景設(shè) 10–60查詢性能優(yōu)先設(shè) 300--roll-sizeMB越大壓縮效果越好性能調(diào)優(yōu)組合建議來自 patterns.md目標(biāo)配置低延遲--roll-interval 10查詢性能--roll-interval 300 --roll-size 100成本最優(yōu)--compression zstd --roll-interval 300六、Pipeline Creation創(chuàng)建不可變 SQL 管道Pipeline 負(fù)責(zé)用 SQL 把 Stream 中的數(shù)據(jù)轉(zhuǎn)換后寫入 Sink命令格式為npx wrangler pipelines create my-pipeline \ --sql INSERT INTO my_sink SELECT * FROM my_stream WHERE event_type purchase6.1 常用 SQL 轉(zhuǎn)換模式管道中的 SQL 遵循INSERT INTO sink SELECT ... FROM stream結(jié)構(gòu)常用模式包括詳見 api.md 與 patterns.md-- ① 過濾事件盡早裁剪減少存儲 INSERT INTO my_sink SELECT * FROM my_stream WHERE event_type purchase AND amount 100 -- ② 只選需要的字段 INSERT INTO my_sink SELECT user_id, event_type, timestamp, amount FROM my_stream -- ③ 轉(zhuǎn)換與富化UPPER / 數(shù)學(xué)運(yùn)算 / CONCAT / CASE WHEN INSERT INTO my_sink SELECT user_id, UPPER(event_type) as event_type, timestamp, amount * 1.1 as amount_with_tax, CONCAT(user_id, _, product_id) as unique_key, CASE WHEN amount 1000 THEN high_value WHEN amount 100 THEN medium_value ELSE low_value END as customer_tier FROM my_stream WHERE event_type IN (purchase, refund)6.2 可用 SQL 函數(shù)速查函數(shù)示例用途UPPER(s)UPPER(event_type)字符串歸一化LOWER(s)LOWER(email)大小寫不敏感匹配CONCAT(...)CONCAT(user_id, _, product_id)生成復(fù)合鍵CASE WHEN ... THEN ... ENDCASE WHEN amount 100 THEN high ELSE low END條件富化CAST(x AS type)CAST(timestamp AS string)類型轉(zhuǎn)換COALESCE(x, y)COALESCE(amount, 0.0)默認(rèn)值兜底數(shù)學(xué)運(yùn)算符amount * 1.1、price / quantity計(jì)算比較運(yùn)算amount 100、status IN (active, pending)過濾CAST支持的字符串類型與 Schema 類型一致string、int32、int64、float32、float64、bool、timestamp。6.3 ?? Pipelines 是不可變的創(chuàng)建后無法修改 SQL只能刪除重建npx wrangler pipelines delete old-pipeline npx wrangler pipelines create new-pipeline --sql ...配套的 SQL 限制包括不支持 JOIN單管道只處理單個(gè)流、不支持窗口函數(shù)、不支持子查詢、無 Schema 演化詳見 gotchas.md。因此官方建議使用版本化命名如events-pipeline-v1將 SQL納入版本控制Schema 演化時(shí)采用雙寫過渡策略創(chuàng)建 v2 流/管道后await Promise.all([env.EVENTS_V1.send([event]), env.EVENTS_V2.send([event])])同時(shí)寫入新舊版本過渡期結(jié)束后刪除舊管道詳見 patterns.md。七、Credentials三類憑據(jù)速查類型所需權(quán)限獲取位置Catalog tokenIceberg Sink 用R2 Admin Read WriteDashboard → R2 → API tokensR2 credentialsRaw Sink 用Object Read Writewrangler r2 bucket create的輸出HTTP ingest token外部寫入用Workers Pipeline SendDashboard → Workers → API tokens安全建議來自 r2-data-catalog 配置文檔Token 應(yīng)通過環(huán)境變量或密鑰管理器保存、絕不硬編碼進(jìn)代碼遵循最小權(quán)限原則查詢引擎用只讀 Token寫入才用讀寫 Token定期輪換 Token每個(gè)應(yīng)用單獨(dú)建 Token 以便追蹤與吊銷。八、查詢落庫數(shù)據(jù)R2 Data Catalog若 Sink 是 Iceberg 表可用 Wrangler 直接對數(shù)據(jù)跑標(biāo)準(zhǔn) SQL含 GROUP BY、JOIN、WHERE、ORDER BY 等export WRANGLER_R2_SQL_AUTH_TOKENYOUR_CATALOG_TOKEN npx wrangler r2 sql query warehouse_name SELECT event_type, COUNT(*) as event_count, SUM(amount) as total_revenue FROM default.my_table WHERE event_type purchase AND timestamp 2025-01-01 GROUP BY event_type ORDER BY total_revenue DESC LIMIT 100九、完整示例一條生產(chǎn)級 ETL 管道的落地全流程配置文檔給出的端到端流程my-bucket需為啟用了 Data Catalog 的桶# 1. 創(chuàng)建并啟用 R2 桶的 Data Catalog npx wrangler r2 bucket create my-bucket npx wrangler r2 bucket catalog enable my-bucket # 2. 用 Schema 文件創(chuàng)建結(jié)構(gòu)化流 npx wrangler pipelines streams create my-stream --schema-file schema.json # 3. 創(chuàng)建 Iceberg Sink按需調(diào)整 --catalog-token 等參數(shù) npx wrangler pipelines sinks create my-sink --type r2-data-catalog --bucket my-bucket ... # 4. 創(chuàng)建不可變 SQL 管道 npx wrangler pipelines create my-pipeline --sql INSERT INTO my_sink SELECT * FROM my_stream # 5. 部署 Worker讓綁定生效 npx wrangler deploy部署前請確認(rèn)已通過npx wrangler whoami完成認(rèn)證未認(rèn)證時(shí)參考 wrangler/auth.md本地用wrangler loginCI/CD 用CLOUDFLARE_API_TOKEN環(huán)境變量。十、調(diào)試清單與常見錯(cuò)誤診斷清單來自 gotchas.md流存在npx wrangler pipelines streams list管道健康npx wrangler pipelines get IDSQL 語法與 Schema 字段匹配添加綁定后已重新部署 Worker已等待 roll interval10–300 秒Accepted 數(shù)量與 Processed 數(shù)量一致無校驗(yàn)靜默丟棄常見錯(cuò)誤對照錯(cuò)誤原因修復(fù)事件不在 R2 中roll interval 未到等待 10–300s檢查roll_intervalSchema 校驗(yàn)失敗類型不匹配、缺必填字段客戶端先行校驗(yàn)限流429單流寫入 5 MB/s批量發(fā)送、申請?zhí)岣哳~度負(fù)載過大413單請求 1 MB拆分為更小的批次無法刪除流仍有 Pipeline 引用先刪除管道Sink 憑據(jù)錯(cuò)誤Token 過期用新憑據(jù)重建 SinkOpen Beta 限制以倉庫文檔為準(zhǔn)每個(gè)賬戶 Streams/Sinks/Pipelines 各 20 個(gè)Payload 上限 1 MB單流攝入速率 5 MB/s事件保留 24 小時(shí)推薦批量大小 100 個(gè)事件。延伸閱讀本指南聚焦配置鏈路若需繼續(xù)深入建議按 Pipelines README 的閱讀順序展開api.md —— 發(fā)送事件、TypeScript 類型、SQL 函數(shù)全參考、HTTP 響應(yīng)碼patterns.md —— Fire-and-forget、Zod 校驗(yàn)、Pipelines Queues 扇出、性能調(diào)優(yōu)、Schema 版本化gotchas.md —— 靜默丟棄、不可變管道、限流與限制r2-data-catalog —— 桶的 Catalog 啟用、PyIceberg 客戶端配置與 Token 權(quán)限細(xì)節(jié)r2 —— R2 桶管理、S3 SDK 接入、生命周期與事件通知【免費(fèi)下載鏈接】skillsSkills Catalog for Codex項(xiàng)目地址: https://gitcode.com/GitHub_Trending/skills4/skills創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考