Cloudflare Pipelines 配置全指南从 Worker 绑定、Streams/Sinks 到不可变 SQL 管道的完整实战手册【免费下载链接】skillsSkills Catalog for Codex项目地址: https://gitcode.com/GitHub_Trending/skills4/skills本指南以 Cloudflare Deploy Skill 中 Pipelines 配置文档 为骨架系统讲解 Cloudflare PipelinesETL 流式平台的完整配置链路在wrangler.jsonc中声明 Worker 绑定、为结构化流定义 Schema、用 CLI 创建 Streams 与两类 SinkIceberg 数据目录 / Parquet 原始存储、创建不可变 SQL 管道并落地到 R2。阅读完成后你将能够从零搭建一条数据源 → Streams → Pipelines(SQL) → Sinks → R2的生产级流式 ETL 管道并掌握 Schema 校验、凭据配置、性能调参与常见故障排查的完整要点。一、Pipelines 是什么一个面向 R2 的流式 ETL 平台Cloudflare Pipelines 是一个用于采集Ingest、转换Transform并加载Load数据到 R2的流式 ETL 平台。其核心架构由三部分组成见 Pipelines READMEData Sources → Streams → Pipelines (SQL) → Sinks → R2 ↑ ↓ ↓ HTTP/Workers Transform Iceberg/Parquet组件职责关键特性Streams事件采集持久的缓冲层HTTP / Workers 写入支持结构化带 Schema 校验与非结构化两种模式Pipelines用 SQL 对流做转换创建后不可变immutable无法修改 SQLSinks将数据写入 R2 目的地提供精确一次exactly-once投递语义常见的落地形态分析管道点击流、遥测、服务器日志、数据仓库ETL 进可查询的 Iceberg 表、事件处理移动端 / IoT 富化、电商分析用户事件、购买、浏览。当前状态为Open Beta需要 Workers Paid 套餐除标准 R2 存储/操作费用外不额外计费以仓库文档声明为准。二、Worker Binding在 wrangler.jsonc 中声明流绑定要让 Worker 代码能够写入流首先需要在wrangler.jsonc中声明Pipelines 绑定。配置文档给出的完整示例// wrangler.jsonc { pipelines: [ { pipeline: STREAM_ID, binding: STREAM } ] }要点解析pipeline字段填的是Stream ID流 ID而不是 Pipeline管道ID——这是最容易踩的坑。通过以下命令获取流 IDnpx wrangler pipelines streams listbinding是你自己在 Worker 代码中使用的环境变量名如STREAM之后在Env类型中声明并调用env.STREAM.send(...)即可写入事件。修改绑定后必须重新部署npx wrangler deploy才会生效若出现env.STREAM is undefined优先检查这两点详见 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 模式不阻塞响应 ctx.waitUntil(env.STREAM.send([event])); return new Response(OK); } } satisfies ExportedHandlerEnv;三、Schema为结构化流定义字段与类型创建结构化流时可以用一个 JSON 文件描述事件结构Pipelines 会据此在写入时做字段级校验。配置文档中的标准 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 } ] }每个字段由三个属性组成属性含义说明name字段名与 SQL 转换中引用的列名一致type字段类型见下方支持类型清单required是否必填true缺失即校验失败false允许缺省支持的类型配置文档原文string、int32、int64、float32、float64、bool、timestamp、json、binary、list、struct⚠️ 重要提示Gotcha结构化流对不合法的事件会静默丢弃——HTTP 返回 200 但事件永远不会出现在 Sink 中详见 gotchas.md。因此强烈建议在客户端用 Zod 先做一次校验获得即时反馈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); // 校验失败会抛异常 await env.STREAM.send([validated]); } catch (e) { // 在此获得即时错误反馈 }四、Stream Setup创建、查询与删除流流的创建有两种模式# 带 Schema结构化流写入时校验 npx wrangler pipelines streams create my-stream --schema-file schema.json # 不带 Schema非结构化流无校验 npx wrangler pipelines streams create my-stream生命周期管理命令# 列出所有流拿到 Stream ID npx wrangler pipelines streams list # 查看单个流详情 npx wrangler pipelines streams get ID # 删除流注意若还有 Pipeline 引用它需先删除管道 npx wrangler pipelines streams delete ID实战提示若删除流时提示失败通常是因为该流仍被某个 Pipeline 引用应先删除管道再删流见 gotchas.md 错误对照表。向流写入事件的两种途径Worker 绑定上文已述env.STREAM.send(events)支持单个对象或数组单次请求上限1 MB单流写入速率上限5 MB/s。HTTP Ingest 端点适用于外部系统/服务端上报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鉴权 Token 需要Workers Pipeline Send权限Dashboard → Workers → API tokens。关键细节HTTP 端点的请求体必须是 JSON 数组而不是单个对象否则会返回 400详见 api.md。五、Sink Configuration两类 R2 落地目标Sink 决定数据最终以什么形态写入 R2。配置文档给出了两种类型选择依据如下来自 README需要直接对数据跑 SQL 查询→ 选R2 Data CatalogIceberg 表具备 ACID 事务、时间旅行time-travel、Schema 演化能力但配置更复杂需要 namespace、table、catalog token仅做文件存储/归档→ 选R2 RawParquet/JSON 文件简单直接但没有内置 SQL 查询配合外部工具Spark / Athena→ 选R2 RawParquet 分区标准格式 分区裁剪提升查询性能但 Schema 兼容性需自行维护。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权限的 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 风格的占位符做 Hive 风格分区目录方便外部查询引擎做分区裁剪。5.3 关键参数速查表配置文档原文OptionValuesGuidance--compressionzstd、snappy、gzipzstd压缩比最佳snappy速度最快--roll-interval秒低延迟场景设 10–60查询性能优先设 300--roll-sizeMB越大压缩效果越好性能调优组合建议来自 patterns.md目标配置低延迟--roll-interval 10查询性能--roll-interval 300 --roll-size 100成本最优--compression zstd --roll-interval 300六、Pipeline Creation创建不可变 SQL 管道Pipeline 负责用 SQL 把 Stream 中的数据转换后写入 Sink命令格式为npx wrangler pipelines create my-pipeline \ --sql INSERT INTO my_sink SELECT * FROM my_stream WHERE event_type purchase6.1 常用 SQL 转换模式管道中的 SQL 遵循INSERT INTO sink SELECT ... FROM stream结构常用模式包括详见 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 -- ③ 转换与富化UPPER / 数学运算 / 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 函数速查函数示例用途UPPER(s)UPPER(event_type)字符串归一化LOWER(s)LOWER(email)大小写不敏感匹配CONCAT(...)CONCAT(user_id, _, product_id)生成复合键CASE WHEN ... THEN ... ENDCASE WHEN amount 100 THEN high ELSE low END条件富化CAST(x AS type)CAST(timestamp AS string)类型转换COALESCE(x, y)COALESCE(amount, 0.0)默认值兜底数学运算符amount * 1.1、price / quantity计算比较运算amount 100、status IN (active, pending)过滤CAST支持的字符串类型与 Schema 类型一致string、int32、int64、float32、float64、bool、timestamp。6.3 ⚠️ Pipelines 是不可变的创建后无法修改 SQL只能删除重建npx wrangler pipelines delete old-pipeline npx wrangler pipelines create new-pipeline --sql ...配套的 SQL 限制包括不支持 JOIN单管道只处理单个流、不支持窗口函数、不支持子查询、无 Schema 演化详见 gotchas.md。因此官方建议使用版本化命名如events-pipeline-v1将 SQL纳入版本控制Schema 演化时采用双写过渡策略创建 v2 流/管道后await Promise.all([env.EVENTS_V1.send([event]), env.EVENTS_V2.send([event])])同时写入新旧版本过渡期结束后删除旧管道详见 patterns.md。七、Credentials三类凭据速查类型所需权限获取位置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 应通过环境变量或密钥管理器保存、绝不硬编码进代码遵循最小权限原则查询引擎用只读 Token写入才用读写 Token定期轮换 Token每个应用单独建 Token 以便追踪与吊销。八、查询落库数据R2 Data Catalog若 Sink 是 Iceberg 表可用 Wrangler 直接对数据跑标准 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九、完整示例一条生产级 ETL 管道的落地全流程配置文档给出的端到端流程my-bucket需为启用了 Data Catalog 的桶# 1. 创建并启用 R2 桶的 Data Catalog npx wrangler r2 bucket create my-bucket npx wrangler r2 bucket catalog enable my-bucket # 2. 用 Schema 文件创建结构化流 npx wrangler pipelines streams create my-stream --schema-file schema.json # 3. 创建 Iceberg Sink按需调整 --catalog-token 等参数 npx wrangler pipelines sinks create my-sink --type r2-data-catalog --bucket my-bucket ... # 4. 创建不可变 SQL 管道 npx wrangler pipelines create my-pipeline --sql INSERT INTO my_sink SELECT * FROM my_stream # 5. 部署 Worker让绑定生效 npx wrangler deploy部署前请确认已通过npx wrangler whoami完成认证未认证时参考 wrangler/auth.md本地用wrangler loginCI/CD 用CLOUDFLARE_API_TOKEN环境变量。十、调试清单与常见错误诊断清单来自 gotchas.md流存在npx wrangler pipelines streams list管道健康npx wrangler pipelines get IDSQL 语法与 Schema 字段匹配添加绑定后已重新部署 Worker已等待 roll interval10–300 秒Accepted 数量与 Processed 数量一致无校验静默丢弃常见错误对照错误原因修复事件不在 R2 中roll interval 未到等待 10–300s检查roll_intervalSchema 校验失败类型不匹配、缺必填字段客户端先行校验限流429单流写入 5 MB/s批量发送、申请提高额度负载过大413单请求 1 MB拆分为更小的批次无法删除流仍有 Pipeline 引用先删除管道Sink 凭据错误Token 过期用新凭据重建 SinkOpen Beta 限制以仓库文档为准每个账户 Streams/Sinks/Pipelines 各 20 个Payload 上限 1 MB单流摄入速率 5 MB/s事件保留 24 小时推荐批量大小 100 个事件。延伸阅读本指南聚焦配置链路若需继续深入建议按 Pipelines README 的阅读顺序展开api.md —— 发送事件、TypeScript 类型、SQL 函数全参考、HTTP 响应码patterns.md —— Fire-and-forget、Zod 校验、Pipelines Queues 扇出、性能调优、Schema 版本化gotchas.md —— 静默丢弃、不可变管道、限流与限制r2-data-catalog —— 桶的 Catalog 启用、PyIceberg 客户端配置与 Token 权限细节r2 —— R2 桶管理、S3 SDK 接入、生命周期与事件通知【免费下载链接】skillsSkills Catalog for Codex项目地址: https://gitcode.com/GitHub_Trending/skills4/skills创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考