管理 pipeline

了解如何：

创建带有 SQL 转换的 pipeline
查看 pipeline 配置和 SQL
在不再需要时删除 pipeline

创建 pipeline

Pipeline 执行定义数据如何从 stream 流向 sink 的 SQL 语句。

仪表板

在 Cloudflare 仪表板中，前往 Basin Pipelines 页面。
前往 Pipelines ↗
选择 Create Pipeline（创建 Pipeline） 启动 pipeline 创建向导。
按照向导配置 stream、sink 和 SQL 转换。

Wrangler CLI

要创建 pipeline，请运行 basin pipelines create 命令：

npx wrangler basin pipelines create my-pipeline --sql "INSERT INTO my_sink SELECT * FROM my_stream"

yarn wrangler basin pipelines create my-pipeline --sql "INSERT INTO my_sink SELECT * FROM my_stream"

pnpm wrangler basin pipelines create my-pipeline --sql "INSERT INTO my_sink SELECT * FROM my_stream"

您也可以从文件提供 SQL：

npx wrangler basin pipelines create my-pipeline --sql-file pipeline.sql

yarn wrangler basin pipelines create my-pipeline --sql-file pipeline.sql

pnpm wrangler basin pipelines create my-pipeline --sql-file pipeline.sql

或者，要运行帮助您配置 stream、sink 和 pipeline 的交互式设置向导，请运行 basin pipelines setup 命令：

npx wrangler basin pipelines setup

yarn wrangler basin pipelines setup

pnpm wrangler basin pipelines setup

SQL 转换

Pipeline 支持用于数据转换的 SQL 语句。有关完整语法、支持的函数和数据类型，请参阅 SQL 参考。

常见模式包括：

基本数据流

将所有数据从 stream 传输到 sink：

INSERT INTO my_sink SELECT * FROM my_stream

过滤事件

根据条件过滤事件：

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

转换数据

对字段应用转换：

INSERT INTO my_sink
SELECT
  user_id,
  UPPER(event_type) as event_type,
  timestamp,
  amount * 1.1 as amount_with_tax
FROM my_stream

将一个 stream 路由到多个表

单个 pipeline 可以运行多个用分号分隔的 INSERT 语句。每个语句从同一 stream 读取并写入不同的 sink，因此您可以根据内容将（"扇出"）一个 stream 中的事件路由到多个表。

这避免了为每个目的地运行单独的 pipeline。每个语句使用自己的 WHERE 子句过滤 stream，并仅投影与该表相关的列。这也可以作为成本优化，因为您将按定价为转换计费一次，而非按语句计费。

INSERT INTO purchases_sink
SELECT user_id, product_id, amount FROM my_stream
WHERE event_type = 'purchase';

INSERT INTO page_views_sink
SELECT user_id, product_id FROM my_stream
WHERE event_type = 'view_product';

有关将实时事件 stream 扇出到五个表的完整示例，请参阅将 stream 扇出到多个 Iceberg 表。

查看 pipeline 配置

仪表板

在 Cloudflare 仪表板中，前往 Basin Pipelines 页面。
选择一个 pipeline 以查看其 SQL 转换、连接的 stream/sink 以及关联指标。

Wrangler CLI

要查看特定 pipeline，请使用 pipeline ID 或 pipeline 名称运行 basin pipelines get 命令：

npx wrangler basin pipelines get <PIPELINE_NAME_OR_ID>

yarn wrangler basin pipelines get <PIPELINE_NAME_OR_ID>

pnpm wrangler basin pipelines get <PIPELINE_NAME_OR_ID>

要列出账户中的所有 pipeline，请运行 basin pipelines list 命令：

npx wrangler basin pipelines list

yarn wrangler basin pipelines list

pnpm wrangler basin pipelines list

删除 pipeline

删除 pipeline 会停止连接的 stream 到 sink 的数据流。

仪表板

在 Cloudflare 仪表板中，前往 Basin Pipelines 页面。
选择要删除的 pipeline。 3. 在 Settings（设置） 选项卡中，选择 Delete（删除）。

Wrangler CLI

要删除 pipeline，请运行 basin pipelines delete 命令：

npx wrangler basin pipelines delete <PIPELINE_ID>

yarn wrangler basin pipelines delete <PIPELINE_ID>

pnpm wrangler basin pipelines delete <PIPELINE_ID>

限制

Pipeline SQL 创建后无法修改。要更改 SQL 转换，必须删除并重新创建 pipeline。