本指南将指导您完成以下操作：
- 创建 pipeline 与数据目录进行身份验证所需的 API 令牌。
- 使用简单的电商 schema 创建第一个 pipeline，写入由 R2 Data Catalog 管理的 Apache Iceberg ↗ 表。
- 通过 HTTP 端点发送示例电商数据。
- 验证存储桶中的数据并使用 R2 SQL 查询。
- 注册 Cloudflare 账户 ↗。
- 安装
Node.js↗。
Node.js 版本管理器
使用 Volta ↗ 或 nvm ↗ 等 Node 版本管理器，以避免权限问题并切换 Node.js 版本。本指南后续将介绍的 Wrangler 需要 Node 版本 16.17.0 或更高。
Pipeline 必须使用具有 catalog 和 R2 权限的 R2 API 令牌 向 R2 Data Catalog 进行身份验证。
-
在 Cloudflare 仪表板中，前往 R2 object storage（R2 对象存储） 页面。
Go to Overview ↗ -
选择 Manage API tokens（管理 API 令牌）。
-
选择 Create Account API token（创建账户 API 令牌）。
-
为 API 令牌命名。
-
在 Permissions（权限） 下，选择 Admin Read & Write（管理员读取和写入） 权限。
-
选择 Create Account API Token（创建账户 API 令牌）。
-
记下 Token value。
首先，创建一个定义电商数据结构的 schema 文件：
创建 schema.json：
{
"fields": [
{
"name": "user_id",
"type": "string",
"required": true
},
{
"name": "event_type",
"type": "string",
"required": true
},
{
"name": "product_id",
"type": "string",
"required": false
},
{
"name": "amount",
"type": "float64",
"required": false
}
]
}使用交互式设置创建写入 R2 Data Catalog 的 pipeline：
npx wrangler pipelines setup按照提示操作：
-
Pipeline name：输入
ecommerce -
Stream configuration：
- 启用 HTTP 端点：
yes - 需要身份验证：
no（为简化起见） - 配置自定义 CORS 来源：
no - Schema 定义：
Load from file - Schema 文件路径：
schema.json（或您的文件路径）
- 启用 HTTP 端点：
-
Sink configuration：
- 目的地类型：
Data Catalog (Iceberg) - 设置模式：
Simple (recommended defaults) - R2 存储桶名称：
pipelines-tutorial（如果不存在会自动创建） - 表名称：
ecommerce - Catalog API 令牌：输入步骤 1 中的令牌
- 目的地类型：
-
Review：确认摘要并选择
Create resources -
SQL transformation：选择
Simple ingestion (SELECT * FROM stream)
高级模式选项
如果在 sink 配置期间选择 Advanced（高级） 而非 Simple（简单），您可以自定义以下附加选项：
- Format：输出文件格式（例如 Parquet）
- Compression：压缩算法（例如 zstd）
- Rolling policy：创建新文件的文件大小阈值（最小 5 MB）和时间间隔（最小 10 秒）
- Credentials：在自动凭据生成和手动输入 R2 凭据之间选择
- Namespace：Data Catalog 命名空间（默认为
default）
设置完成后，命令会输出 Wrangler 文件的配置片段、带有示例数据的 Worker 绑定示例，以及 HTTP 端点的 curl 命令。记下 HTTP 端点 URL 和 pipelines 配置，以便在后续步骤中使用。
您也可以使用 --name 标志预设 pipeline 名称：
npx wrangler pipelines setup --name ecommerce-
在 Cloudflare 仪表板中，前往 R2 object storage（R2 对象存储）。
Go to Overview ↗ -
选择 Create bucket（创建存储桶） 并输入存储桶名称：
pipelines-tutorial。 -
选择 Create bucket（创建存储桶）。
-
选择该存储桶，切换到 Settings（设置） 选项卡，向下滚动到 R2 Data Catalog（R2 数据目录），然后选择 Enable（启用）。
-
启用后，记下 Catalog URI（目录 URI） 和 Warehouse name（仓库名称）。
-
前往 Pipelines > Pipelines（管道）。
Go to Pipelines ↗ -
选择 Create Pipeline（创建 Pipeline）。
-
Connect to a Stream（连接到 Stream）：
- Pipeline name（管道名称）：
ecommerce - Enable HTTP endpoint for sending data（启用用于发送数据的 HTTP 端点）：已启用
- HTTP authentication（HTTP 身份验证）：已禁用（默认）
- 选择 Next（下一步）
- Pipeline name（管道名称）：
-
Define Input Schema（定义输入架构）：
- 选择 JSON editor（JSON 编辑器）
- 复制 schema：
{ "fields": [ { "name": "user_id", "type": "string", "required": true }, { "name": "event_type", "type": "string", "required": true }, { "name": "product_id", "type": "string", "required": false }, { "name": "amount", "type": "float64", "required": false } ] } - 选择 Next（下一步）
-
Define Sink（定义接收器）：
- 选择 R2 存储桶：
pipelines-tutorial - 存储类型：R2 Data Catalog（R2 数据目录）
- Namespace（命名空间）：
default - Table name（表名称）：
ecommerce - Advanced Settings（高级设置）：将 Maximum Time Interval（最大时间间隔） 更改为
10 seconds - 选择 Next（下一步）
- 选择 R2 存储桶：
-
Credentials（凭据）：
- 禁用 Automatically create an Account API token for your sink（自动为接收器创建账户 API 令牌）
- 输入步骤 1 中的 Catalog Token（目录令牌）
- 选择 Next（下一步）
-
Pipeline Definition（管道定义）：
- 保留默认 SQL 查询：
INSERT INTO ecommerce_sink SELECT * FROM ecommerce_stream; - 选择 Create Pipeline（创建 Pipeline）
- 保留默认 SQL 查询：
-
pipeline 创建后，记下 Stream ID（流 ID） 以供下一步使用。
向 pipeline 的 HTTP 端点发送电商事件：
curl -X POST https://{stream-id}.ingest.cloudflare.com \
-H "Content-Type: application/json" \
-d '[
{
"user_id": "user_12345",
"event_type": "purchase",
"product_id": "widget-001",
"amount": 29.99
},
{
"user_id": "user_67890",
"event_type": "view_product",
"product_id": "widget-002"
},
{
"user_id": "user_12345",
"event_type": "add_to_cart",
"product_id": "widget-003",
"amount": 15.50
}
]'将 {stream-id} 替换为 pipeline 设置中的实际 stream 端点。
-
在 Cloudflare 仪表板中，前往 R2 object storage（R2 对象存储） 页面。
-
选择存储桶：
pipelines-tutorial。 -
您应该看到 pipeline 创建的 Iceberg 元数据文件和数据文件。如果在存储桶中看不到任何文件，请等待几分钟后重试。
-
数据以 Apache Iceberg 格式组织，元数据跟踪表版本。
设置环境以使用 R2 SQL：
export WRANGLER_R2_SQL_AUTH_TOKEN=YOUR_API_TOKEN或创建包含以下内容的 .env 文件：
WRANGLER_R2_SQL_AUTH_TOKEN=YOUR_API_TOKEN其中 YOUR_API_TOKEN 是您在步骤 1 中创建的令牌。有关设置环境变量的更多信息，请参阅 Wrangler 系统环境变量。
查询数据：
npx wrangler r2 sql query "YOUR_WAREHOUSE_NAME" "
SELECT
user_id,
event_type,
product_id,
amount
FROM default.ecommerce
WHERE event_type = 'purchase'
LIMIT 10"将 YOUR_WAREHOUSE_NAME 替换为 pipeline 设置期间记下的 warehouse 名称。您可以在 Cloudflare 仪表板的 R2 object storage（R2 对象存储） > 您的存储桶 > Settings（设置） > R2 Data Catalog（R2 数据目录） 下找到它。
您还可以使用任何支持 Apache Iceberg 的引擎查询此表。有关将其他引擎连接到 R2 Data Catalog 的更多信息，请参阅连接到 Iceberg 引擎。