快速入门 · Basin SQL 文档

本指南将指导您完成以下操作：

创建您的第一个 R2 存储桶并启用其数据目录。
创建管道在数据目录进行身份验证所需的 API 令牌。
使用简单的电子商务架构创建您的第一个管道，该管道可写入由 Basin Catalog 管理的 Apache Iceberg ↗ 表。
通过 HTTP 端点发送示例电子商务数据。
验证您的存储桶中的数据并使用 Basin SQL 查询它。

先决条件

注册 Cloudflare 账户 ↗。
安装 Node.js ↗。

Node.js 版本管理器

使用 Vite+ (vp env) ↗ 安装和管理 Node.js 版本。

有关支持的 Node.js 版本，请参阅 Wrangler 系统要求。

1. 创建 R2 存储桶

如果尚未登录，请运行：
npx wrangler login
yarn wrangler login
pnpm wrangler login
创建 R2 存储桶：
npx wrangler r2 bucket create pipelines-tutorial
yarn wrangler r2 bucket create pipelines-tutorial
pnpm wrangler r2 bucket create pipelines-tutorial

在 Cloudflare 仪表板中，转到 R2 object storage（R2 对象存储） 页面。
前往 Overview（概览） ↗
选择 Create bucket（创建存储桶）。
输入存储桶名称：pipelines-tutorial
选择 Create bucket（创建存储桶）。

2. 启用 Basin Catalog

在您的 R2 存储桶上启用目录：

npx wrangler basin catalog enable pipelines-tutorial

yarn wrangler basin catalog enable pipelines-tutorial

pnpm wrangler basin catalog enable pipelines-tutorial

当您运行此命令时，请记下 "Warehouse" 和 "Catalog URI"。稍后您将需要这些内容。

在 Cloudflare 仪表板中，转到 R2 object storage（R2 对象存储） 页面。
前往 Overview（概览） ↗
选择存储桶：pipelines-tutorial。
切换到 Settings（设置） 选项卡，向下滚动到 Basin Catalog，然后选择 Enable（启用）。
启用后，请记下 Catalog URI 和 Warehouse name。

3. 创建 API 令牌

Basin Pipelines 必须使用具有目录和 R2 权限的 R2 API 令牌对 Basin Catalog 进行身份验证。

在 Cloudflare 仪表板中，转到 R2 object storage（R2 对象存储） 页面。
前往 Overview（概览） ↗
选择 Manage API tokens（管理 API 令牌）。
选择 Create Account API token（创建账户 API 令牌）。
为您的 API 令牌命名。
在 Permissions（权限） 下，选择 Admin Read & Write（管理员读写） 权限。
选择 Create Account API Token（创建账户 API 令牌）。
记下 Token value。

4. 创建管道

首先，创建一个定义您的电子商务数据结构的架构文件：

创建 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
		}
	]
}

使用交互式设置创建可写入 Basin Catalog 的管道：

npx wrangler basin pipelines setup

yarn wrangler basin pipelines setup

pnpm wrangler basin pipelines setup

按照提示操作：

Pipeline name：输入 ecommerce
Stream configuration：
- Enable HTTP endpoint: yes
- Require authentication: no （为简单起见）
- Configure custom CORS origins: no
- Schema definition: Load from file
- Schema file path: schema.json （或您的文件路径）
Sink configuration：
- Destination type: Data Catalog Table
- R2 bucket name: pipelines-tutorial
- Namespace: default
- Table name: ecommerce
- Catalog API token: 输入第 3 步中的令牌
- Compression: zstd
- Roll file when size reaches (MB): 100
- Roll file when time reaches (seconds): 10 （为了在本教程中更快地看到数据）
SQL transformation：选择 Use simple ingestion query 以使用：
```
INSERT INTO ecommerce_sink SELECT * FROM ecommerce_stream
```

设置完成后，请记下最终输出中显示的 HTTP 端点 URL。

在 Cloudflare 仪表板中，转到 Basin Pipelines > Pipelines。
前往 Pipelines ↗
选择 Create Pipeline（创建管道）。
Connect to a Stream（连接到 Stream）：
- Pipeline name（管道名称）：ecommerce
- Enable HTTP endpoint for sending data（启用用于发送数据的 HTTP 端点）：已启用
- HTTP authentication（HTTP 身份验证）：已禁用（默认）
- 选择 Next（下一步）

Define Input Schema（定义输入架构）：

选择 JSON editor（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": "f64",
			"required": false
		}
	]
}

选择 Next（下一步）

Define Sink（定义接收器）：
- 选择您的 R2 存储桶：pipelines-tutorial
- Storage type（存储类型）：Basin Catalog
- Namespace（命名空间）：default
- Table name（表名称）：ecommerce
- Advanced Settings（高级设置）：将 Maximum Time Interval（最大时间间隔） 更改为 10 seconds
- 选择 Next（下一步）
Credentials（凭据）：
- 禁用 Automatically create an Account API token for your sink（自动为接收器创建账户 API 令牌）
- 输入第 3 步中的 Catalog Token（目录令牌）
- 选择 Next（下一步）
Pipeline Definition（管道定义）：
- 保留默认的 SQL 查询：
```
INSERT INTO ecommerce_sink SELECT * FROM ecommerce_stream;
```
- 选择 Create Pipeline（创建管道）
管道创建后，请记下 Stream ID（流 ID） 以备下一步使用。

5. 发送示例数据

将电子商务事件发送到您的管道 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} 替换为管道设置中的实际流端点。

6. 验证存储桶中的数据

在 Cloudflare 仪表板中，转到 R2 object storage（R2 对象存储） 页面。
选择您的存储桶：pipelines-tutorial。
确认管道已创建 Iceberg 元数据文件和数据文件。如果文件尚未出现，请等待几分钟后重试。
数据按 Apache Iceberg 格式组织，其中包含用于跟踪表版本的元数据。

7. 使用 Basin SQL 查询您的数据

设置您的环境以使用 Basin SQL：

export WRANGLER_BASIN_SQL_AUTH_TOKEN=YOUR_API_TOKEN

或者创建一个包含以下内容的 .env 文件：

WRANGLER_BASIN_SQL_AUTH_TOKEN=YOUR_API_TOKEN

其中 YOUR_API_TOKEN 是您在第 3 步中创建的令牌。有关设置环境变量的更多信息，请参阅 Wrangler 系统环境变量。

查询您的数据：

npx wrangler basin sql query "YOUR_WAREHOUSE_NAME" "SELECT user_id, event_type, product_id, amount FROM default.ecommerce WHERE event_type = 'purchase' LIMIT 10"

yarn wrangler basin sql query "YOUR_WAREHOUSE_NAME" "SELECT user_id, event_type, product_id, amount FROM default.ecommerce WHERE event_type = 'purchase' LIMIT 10"

pnpm wrangler basin 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 替换为第 2 步中的仓库名称。

您还可以使用任何支持 Apache Iceberg 的引擎查询该表。要将其他引擎连接到 Basin Catalog，请参阅连接到 Iceberg 引擎。

了解更多

管理目录

在您的存储桶上启用或禁用 Basin Catalog，检索配置详细信息，并对您的 Iceberg 引擎进行身份验证。

尝试另一个示例

有关设置简单的欺诈检测数据管道并在 Python 中为其生成事件的详细教程。

Basin Pipelines

了解 SQL 转换和管道配置。