企业官网建设流程全解析

Cloudflare R2 SQL 实战指南：用 Wrangler 与 HTTP API 查询 Apache Iceberg 表

【免费下载链接】skillsSkills Catalog for Codex项目地址: https://gitcode.com/GitHub_Trending/skills4/skills

R2 SQL 是 Cloudflare 推出的无服务器分布式分析查询引擎，它让你直接对存储在 R2 Data Catalog 中的 Apache Iceberg 表执行标准 SQL，无需管理任何集群或基础设施。本文以本仓库 skills/.curated/cloudflare-deploy/references/r2-sql 系列文档为主体，完整覆盖从启用 Catalog、创建 API Token 到编写聚合查询、规避 SQL 限制的端到端实战流程。读完本文，你将掌握如何用 Wrangler CLI 与 HTTP API 完成日志分析、欺诈检测、BI 报表等典型分析任务，并理解其多层级裁剪与并行执行的底层原理。

R2 SQL 是什么

R2 SQL 是 Cloudflare 面向 R2 Data Catalog 中 Apache Iceberg 表的无服务器分布式分析查询引擎，核心特性包括：

• Serverless（无服务器）：无需管理集群，也没有需要运维的基础设施；
• Distributed（分布式）：利用 Cloudflare 全球网络并行执行查询；
• SQL 接口：使用熟悉的 SQL 语法编写分析查询；
• 零出口费用（Zero egress fees）：从任何云/区域查询数据均不产生数据传输成本；
• Open beta（开放测试）：测试期间免费（仅需支付标准 R2 存储费用）。

什么是 Apache Iceberg

Iceberg 是面向对象存储中大规模分析数据集的开源表格式，也是 R2 Data Catalog 的基础：

• ACID 事务：安全的并发读写；
• 元数据优化：无需全表扫描即可快速查询；
• Schema 演化：无需重写即可增删改列；
• 分区（Partitioning）：组织数据以实现高效的裁剪（pruning）。

何时该用 / 不该用 R2 SQL

适合使用 R2 SQL 的场景：

• 日志分析：用 WHERE 过滤与聚合查询应用/系统日志；
• BI 看板：从大规模分析数据集生成报表；
• 欺诈检测：用 GROUP BY / HAVING 分析交易模式；
• 多云分析：跨云查询数据且无出口费用；
• 临时探索：通过 Wrangler CLI 对 Iceberg 表直接跑 SQL。

不建议使用 R2 SQL 的场景：

• Workers/Pages 运行时：R2 SQL 没有 Workers binding，需从外部系统调用 HTTP API（详见下文“关键限制”）；
• 实时查询（<100ms）：它面向分析型批量查询而非 OLTP；
• 复杂 JOIN/CTE：当前 SQL 功能集有限（不支持 JOIN、子查询、CTE）；
• 小数据集（<1GB）：搭建成本不值得。

决策树：如何查询 R2 数据

Do you need to query structured data in R2? ├─ YES, data is in Iceberg tables │ ├─ Need SQL interface? → Use R2 SQL (this reference) │ ├─ Need Python API? → See r2-data-catalog reference (PyIceberg) │ └─ Need other engine? → See r2-data-catalog reference (Spark, Trino, etc.) │ ├─ YES, but not in Iceberg format │ ├─ Streaming data? → Use Pipelines to write to Data Catalog, then R2 SQL │ └─ Static files? → Use PyIceberg to create Iceberg tables, then R2 SQL │ └─ NO, just need object storage → Use R2 reference (not R2 SQL)

若数据尚未组织成 Iceberg 表，可以先通过 Pipelines 将流式数据写入 Data Catalog，或通过 PyIceberg（r2-data-catalog 参考） 创建 Iceberg 表，之后再用 R2 SQL 查询。

架构原理：从查询规划到并行执行

查询规划器（Query Planner）

• 自上而下的元数据侦查 + 多层裁剪：支持分区级（partition-level）、列级（column-level）与行组级（row-group）裁剪，跳过不必要的数据；
• 流式管道：执行在规划完成前就已启动，减少首结果延迟；
• LIMIT 提前终止：结果集一旦完整即停止计算。

查询执行（Query Execution）

• Coordinator 分发：协调器将工作分发给 Cloudflare 网络上的 workers 并行执行；
• Apache DataFusion：worker 使用开源 DataFusion 引擎做并行查询；
• Parquet 列裁剪：只读取查询所需的列；
• R2 范围读取（ranged reads）：按需读取，提升 IO 效率。

聚合策略（Aggregation Strategies）

• Scatter-gather：用于简单聚合（SUM、COUNT、AVG）；
• Shuffling：对聚合结果做 ORDER BY / HAVING 时，通过哈希分区重组数据。

快速开始

完成端到端查询只需四步：

# 1. 在 bucket 上启用 R2 Data Catalog npx wrangler r2 bucket catalog enable my-bucket # 2. 创建 API token（Admin Read & Write） # Dashboard: R2 → Manage API tokens → Create API token # 3. 设置环境变量 export WRANGLER_R2_SQL_AUTH_TOKEN=<your-token> # 4. 运行查询 npx wrangler r2 sql query "my-bucket" "SELECT * FROM default.my_table LIMIT 10"

配置指南

完整配置细节见 configuration.md。

前置条件

• 已启用 Data Catalog 的 R2 bucket；
• 具备 R2 权限的 API token；
• 已安装 Wrangler CLI（用于 CLI 查询）。

启用 R2 Data Catalog

R2 SQL 查询的是 R2 Data Catalog 中的 Apache Iceberg 表，必须先在 bucket 上启用 Catalog。

通过 Wrangler CLI：

npx wrangler r2 bucket catalog enable <bucket-name>

输出包含两项关键信息：

• Warehouse name：通常与 bucket 名一致；
• Catalog URI：REST 目录操作的端点。

示例输出：

Catalog enabled successfully Warehouse: my-bucket Catalog URI: https://abc123.r2.cloudflarestorage.com/iceberg/my-bucket

通过 Dashboard：

1. 进入R2 Object Storage→ 选择 bucket；
2. 点击Settings标签页；
3. 滚动到R2 Data Catalog区域；
4. 点击Enable；
5. 记下Catalog URI与Warehouse名称。

注意：启用 Catalog 会在 bucket 中创建元数据目录，但不会修改已有对象。

创建 API Token

所需权限：R2 Admin Read & Write（已包含 R2 SQL Read 权限）。

Dashboard 操作路径：R2 Object Storage →Manage API tokens（右上角）→Create API token→ 选择Admin Read & Write权限 → 创建后立即复制 token 值（仅展示一次）。

权限范围对照：

注意：R2 SQL Read 权限目前还无法通过 Dashboard 创建，请使用 Admin Read & Write。

配置环境变量

Wrangler CLI 方式，二选一：

export WRANGLER_R2_SQL_AUTH_TOKEN=<your-token>

或创建.env文件（Wrangler 运行命令时会自动加载）：

WRANGLER_R2_SQL_AUTH_TOKEN=<your-token>

HTTP API 方式（非 Wrangler 的程序化访问），将 token 放入 Authorization 头：

curl -X POST https://api.cloudflare.com/client/v4/accounts/{account_id}/r2/sql/query \ -H "Authorization: Bearer <your-token>" \ -H "Content-Type: application/json" \ -d '{ "warehouse": "my-bucket", "query": "SELECT * FROM default.my_table LIMIT 10" }'

注：HTTP API 端点 URL 可能变化，以 patterns.md 的 HTTP API Query 小节 为准。

验证配置

用系统表（system tables）测试连接：

# 列出命名空间 npx wrangler r2 sql query "my-bucket" "SHOW DATABASES" # 列出命名空间中的表 npx wrangler r2 sql query "my-bucket" "SHOW TABLES IN default"

成功时返回 JSON 数组结果。

配置故障排查

SQL 语法与 API 参考

完整语法与类型说明见 api.md。

语法骨架

SELECT column_list | aggregation_function FROM [namespace.]table_name WHERE conditions [GROUP BY column_list] [HAVING conditions] [ORDER BY column | aggregation_function [DESC | ASC]] [LIMIT number]

Schema 发现

SHOW DATABASES; -- 列出命名空间 SHOW NAMESPACES; -- SHOW DATABASES 的别名 SHOW SCHEMAS; -- SHOW DATABASES 的别名 SHOW TABLES IN namespace; -- 列出命名空间中的表 DESCRIBE namespace.table; -- 显示表 schema 与分区键

SELECT 子句

-- 全部列 SELECT * FROM logs.http_requests; -- 指定列 SELECT user_id, timestamp, status FROM logs.http_requests;

限制：不支持列别名、表达式与嵌套列访问。

WHERE 子句操作符

用括号控制优先级，例如：(status = 404 OR status = 500) AND method = 'POST'。

聚合函数

-- GROUP BY 搭配多聚合 SELECT region, COUNT(*), SUM(amount), AVG(amount) FROM sales.transactions WHERE sale_date >= '2024-01-01' GROUP BY region;

HAVING 子句

在 GROUP BY 之后过滤聚合结果：

SELECT category, SUM(amount) FROM sales.transactions GROUP BY category HAVING SUM(amount) > 10000;

ORDER BY 子句

支持两类排序目标：

• 分区键列：始终支持；
• 聚合函数：通过 shuffle 策略支持。

-- 按分区键排序 SELECT * FROM logs.requests ORDER BY timestamp DESC LIMIT 100; -- 按聚合排序（须重复书写函数，不支持别名） SELECT region, SUM(amount) FROM sales.transactions GROUP BY region ORDER BY SUM(amount) DESC;

限制：不能按非分区列排序，详见 gotchas.md 的 ORDER BY 限制。

LIMIT 子句

务必显式使用 LIMIT，以启用提前终止优化。

数据类型与字面量

类型安全要点：字符串用单引号；时间戳必须为 RFC3339 且包含时区；日期用 ISO 8601（YYYY-MM-DD）；不存在隐式类型转换。

-- 正确写法 WHERE status = 200 AND method = 'GET' AND timestamp > '2025-01-01T00:00:00Z' -- 错误写法 WHERE status = '200' -- 字符串替代了整数 WHERE timestamp > '2025-01-01' -- 缺少时间/时区 WHERE method = GET -- 字符串未加引号

查询结果格式

返回 JSON 对象数组：

[ {"user_id": "user_123", "timestamp": "2025-01-15T10:30:00Z", "status": 200}, {"user_id": "user_456", "timestamp": "2025-01-15T10:31:00Z", "status": 404} ]

实战模式：CLI、HTTP API 与生态集成

Wrangler CLI 查询

# 基础查询 npx wrangler r2 sql query "my-bucket" "SELECT * FROM default.logs LIMIT 10" # 多行查询 npx wrangler r2 sql query "my-bucket" " SELECT status, COUNT(*), AVG(response_time) FROM logs.http_requests WHERE timestamp >= '2025-01-01T00:00:00Z' GROUP BY status ORDER BY COUNT(*) DESC LIMIT 100 " # 使用环境变量传递 warehouse export R2_SQL_WAREHOUSE="my-bucket" npx wrangler r2 sql query "$R2_SQL_WAREHOUSE" "SELECT * FROM default.logs"

HTTP API 查询

面向外部系统的程序化访问（注意：不能从 Workers 内调用）：

curl -X POST https://api.cloudflare.com/client/v4/accounts/{account_id}/r2/sql/query \ -H "Authorization: Bearer <your-token>" \ -H "Content-Type: application/json" \ -d '{ "warehouse": "my-bucket", "query": "SELECT * FROM default.my_table WHERE status = 200 LIMIT 100" }'

响应示例：

{ "success": true, "result": [{"user_id": "user_123", "timestamp": "2025-01-15T10:30:00Z", "status": 200}], "errors": [] }

Pipelines 流式集成

先通过 Pipelines 把流数据写入 Iceberg 表，再用 R2 SQL 查询：

# 搭建 pipeline（选择 Data Catalog Table 作为目标） npx wrangler pipelines setup # 关键设置： # - Destination: Data Catalog Table # - Compression: zstd（推荐） # - Roll file time: 生产 300+ 秒，开发 10 秒 # 向 pipeline 发送数据 curl -X POST https://{stream-id}.ingest.cloudflare.com \ -H "Content-Type: application/json" \ -d '[{"user_id": "user_123", "event_type": "purchase", "timestamp": "2025-01-15T10:30:00Z", "amount": 29.99}]' # 等待滚动间隔后查询已入库数据 npx wrangler r2 sql query "my-bucket" " SELECT event_type, COUNT(*), SUM(amount) FROM default.events WHERE timestamp >= '2025-01-15T00:00:00Z' GROUP BY event_type "

详细搭建过程见 pipelines/patterns.md。

PyIceberg 集成

用 PyIceberg 创建并填充 Iceberg 表，再用 R2 SQL 查询：

from pyiceberg.catalog.rest import RestCatalog import pyarrow as pa import pandas as pd # 配置 catalog catalog = RestCatalog( name="my_catalog", warehouse="my-bucket", uri="https://<account-id>.r2.cloudflarestorage.com/iceberg/my-bucket", token="<your-token>", ) catalog.create_namespace_if_not_exists("analytics") # 创建表 schema = pa.schema([ pa.field("user_id", pa.string(), nullable=False), pa.field("event_time", pa.timestamp("us", tz="UTC"), nullable=False), pa.field("page_views", pa.int64(), nullable=False), ]) table = catalog.create_table(("analytics", "user_metrics"), schema=schema) # 追加数据 df = pd.DataFrame({ "user_id": ["user_1", "user_2"], "event_time": pd.to_datetime(["2025-01-15 10:00:00", "2025-01-15 11:00:00"], utc=True), "page_views": [10, 25], }) table.append(pa.Table.from_pandas(df, schema=schema))

查询：

npx wrangler r2 sql query "my-bucket" " SELECT user_id, SUM(page_views) FROM analytics.user_metrics WHERE event_time >= '2025-01-15T00:00:00Z' GROUP BY user_id "

更多高级模式见 r2-data-catalog/patterns.md。

连接外部引擎

R2 Data Catalog 暴露 Iceberg REST API，可连接 Spark、Snowflake、Trino、DuckDB 等引擎：

// Apache Spark 示例 val spark = SparkSession.builder() .config("spark.sql.catalog.my_catalog", "org.apache.iceberg.spark.SparkCatalog") .config("spark.sql.catalog.my_catalog.catalog-impl", "org.apache.iceberg.rest.RESTCatalog") .config("spark.sql.catalog.my_catalog.uri", "https://<account-id>.r2.cloudflarestorage.com/iceberg/my-bucket") .config("spark.sql.catalog.my_catalog.token", "<token>") .getOrCreate() spark.sql("SELECT * FROM my_catalog.default.my_table LIMIT 10").show()

典型业务用例

日志分析：

-- 按端点统计错误率 SELECT path, COUNT(*), SUM(CASE WHEN status >= 400 THEN 1 ELSE 0 END) as errors FROM logs.http_requests WHERE timestamp BETWEEN '2025-01-01T00:00:00Z' AND '2025-01-31T23:59:59Z' GROUP BY path ORDER BY errors DESC LIMIT 20; -- 响应时间统计 SELECT method, MIN(response_time_ms), AVG(response_time_ms), MAX(response_time_ms) FROM logs.http_requests WHERE timestamp >= '2025-01-15T00:00:00Z' GROUP BY method; -- 按状态码统计流量 SELECT status, COUNT(*) FROM logs.http_requests WHERE timestamp >= '2025-01-15T00:00:00Z' AND method = 'GET' GROUP BY status ORDER BY COUNT(*) DESC;

欺诈检测：

-- 大额交易 SELECT location, COUNT(*), SUM(amount), AVG(amount) FROM fraud.transactions WHERE transaction_timestamp >= '2025-01-01T00:00:00Z' AND amount > 1000.0 GROUP BY location ORDER BY SUM(amount) DESC LIMIT 20; -- 标记交易 SELECT merchant_category, COUNT(*), AVG(amount) FROM fraud.transactions WHERE is_fraud_flag = true AND transaction_timestamp >= '2025-01-01T00:00:00Z' GROUP BY merchant_category HAVING COUNT(*) > 10 ORDER BY COUNT(*) DESC;

业务智能（BI）：

-- 按部门统计销售额 SELECT department, SUM(revenue), AVG(revenue), COUNT(*) FROM sales.transactions WHERE sale_date >= '2024-01-01' GROUP BY department ORDER BY SUM(revenue) DESC LIMIT 10; -- 产品表现 SELECT category, COUNT(DISTINCT product_id), SUM(units_sold), SUM(revenue) FROM sales.product_sales WHERE sale_date BETWEEN '2024-10-01' AND '2024-12-31' GROUP BY category ORDER BY SUM(revenue) DESC;

关键限制与陷阱

完整说明见 gotchas.md。

关键限制：没有 Workers Binding

无法从 Workers/Pages 代码中直接调用 R2 SQL，不存在 binding：

// 以下写法不存在 export default { async fetch(request, env) { const result = await env.R2_SQL.query("SELECT * FROM table"); // 不可行 return Response.json(result); } };

替代方案：

• 从外部系统调用 HTTP API（不能从 Workers 内）；
• 通过 r2-data-catalog 的 REST API 使用 PyIceberg/Spark；
• Workers 场景改用 D1 或外部数据库。

ORDER BY 限制

只能按以下两类排序：分区键列（始终支持）、聚合函数（shuffle 策略支持）。普通非分区列不能排序。

-- 合法：按分区键排序 SELECT * FROM logs.requests ORDER BY timestamp DESC LIMIT 100; -- 合法：按聚合排序 SELECT region, SUM(amount) FROM sales.transactions GROUP BY region ORDER BY SUM(amount) DESC; -- 非法：按非分区列排序 SELECT * FROM logs.requests ORDER BY user_id; -- 非法：按别名排序（必须重复写聚合函数） SELECT region, SUM(amount) as total FROM sales.transactions GROUP BY region ORDER BY total; -- 应写 ORDER BY SUM(amount)

用DESCRIBE namespace.table_name检查分区规范。

SQL 功能限制一览

变通方案：无 JOIN 时反规范化数据或使用 Spark/PyIceberg；无子查询时拆分为多次查询；无别名时接受自动生成的列名并在应用层转换。

常见错误与解法

性能问题

慢查询的成因：分区过多、LIMIT 过大、缺少过滤条件、文件过小。

-- 慢：无过滤 SELECT * FROM logs.requests LIMIT 10000; -- 快：按分区键过滤 SELECT * FROM logs.requests WHERE timestamp >= '2025-01-15T00:00:00Z' AND timestamp < '2025-01-16T00:00:00Z' LIMIT 1000; -- 更快：多重过滤 SELECT * FROM logs.requests WHERE timestamp >= '2025-01-15T00:00:00Z' AND status = 404 AND method = 'GET' LIMIT 1000;

文件优化：目标 Parquet 文件大小 100–500MB（压缩后）；Pipelines 滚动间隔生产 300+ 秒、开发 10 秒；定期做 compaction 合并小文件。

查询超时：添加更严格的 WHERE 过滤、缩小时间范围、查询更小的区间。

-- 会超时：全年聚合 SELECT status, COUNT(*) FROM logs.requests WHERE timestamp >= '2024-01-01T00:00:00Z' GROUP BY status; -- 更快：按月聚合 SELECT status, COUNT(*) FROM logs.requests WHERE timestamp >= '2025-01-01T00:00:00Z' AND timestamp < '2025-02-01T00:00:00Z' GROUP BY status;

最佳实践与调试清单

分区策略

• 时序数据：按 timestamp 做 day/hour 分区；
• 地理数据：按 region/country 分区；
• 避免：高基数键（如 user_id）、超过 10,000 个分区。

from pyiceberg.partitioning import PartitionSpec, PartitionField from pyiceberg.transforms import DayTransform PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=DayTransform(), name="day"))

查询书写

• 始终加 LIMIT，触发提前终止；
• 先过滤分区键，实现高效裁剪；
• 用 AND 组合多个过滤条件，裁剪更彻底：

WHERE timestamp >= '2025-01-15T00:00:00Z' AND status = 404 AND method = 'GET' LIMIT 100

类型安全

• 字符串加单引号：'GET'而非GET；
• 时间戳用 RFC3339：'2025-01-01T00:00:00Z'而非'2025-01-01'；
• 日期用 ISO：'2025-01-15'而非'01/15/2025'。

数据组织

• Pipelines：开发roll_file_time: 10，生产roll_file_time: 300+；
• 压缩：使用zstd；
• 维护：对小文件做 compaction，过期旧快照。

调试检查清单

1. npx wrangler r2 bucket catalog enable <bucket>—— 验证 Catalog；
2. echo $WRANGLER_R2_SQL_AUTH_TOKEN—— 检查 token；
3. SHOW DATABASES—— 列出命名空间；
4. SHOW TABLES IN namespace—— 列出表；
5. DESCRIBE namespace.table—— 检查 schema；
6. SELECT COUNT(*) FROM namespace.table—— 验证数据；
7. SELECT * FROM namespace.table LIMIT 10—— 测试简单查询；
8. 逐步添加过滤条件。

相关参考

• R2 SQL 配置指南 —— 启用 Catalog、创建 token、环境配置
• R2 SQL API 参考 —— SQL 语法、函数、操作符、数据类型
• R2 SQL 实战模式 —— Wrangler CLI、HTTP API、Pipelines、PyIceberg 示例
• R2 SQL 限制与排障 —— 限制、常见错误、性能建议
• r2-data-catalog 参考 —— PyIceberg、REST API、外部引擎接入
• pipelines 参考 —— 流式写入 Iceberg 表
• r2 参考 —— R2 对象存储基础
• cloudflare-deploy 技能总览 —— 各 Cloudflare 产品参考索引（存储决策树中 R2 SQL 的定位见 "I need to store data" 一节）

【免费下载链接】skillsSkills Catalog for Codex项目地址: https://gitcode.com/GitHub_Trending/skills4/skills