SELECT 语句

SELECT 语句用于在 Cloudflare Pipelines 中转换数据。一般形式为：

[WITH with_query [, ...]]
SELECT select_expr [, ...]
FROM from_item
[WHERE condition]

Pipeline 运行一个或多个 INSERT INTO sink SELECT ... FROM stream 语句。要从同一 pipeline 写入多个 sink，请用分号分隔语句。参见多条语句。

WITH 子句

WITH 子句允许您定义可在主查询中引用的命名子查询。这可以通过分解复杂转换来提高查询可读性。

语法：

WITH query_name AS (subquery) [, ...]

简单示例：

WITH filtered_events AS
    (SELECT user_id, event_type, amount
        FROM user_events WHERE amount > 50)
SELECT user_id, amount * 1.1 as amount_with_tax
FROM filtered_events
WHERE event_type = 'purchase';

SELECT 子句

SELECT 子句是以逗号分隔的表达式列表，可选别名。列名必须唯一。

SELECT select_expr [, ...]

示例：

-- Select specific columns
SELECT user_id, event_type, amount FROM events

-- Use expressions and aliases
SELECT
    user_id,
    amount * 1.1 as amount_with_tax,
    UPPER(event_type) as event_type_upper
FROM events

-- Select all columns
SELECT * FROM events

FROM 子句

FROM 子句指定查询的数据源。它可以是表名或子查询。表名可以是 stream 名称或在 WITH 子句中创建的表。

FROM from_item

表可以指定别名：

SELECT e.user_id, e.amount
FROM user_events e
WHERE e.event_type = 'purchase'

WHERE 子句

WHERE 子句使用布尔条件过滤数据。谓词应用于输入行。

WHERE condition

示例：

-- Filter by field value
SELECT * FROM events WHERE event_type = 'purchase'

-- Multiple conditions
SELECT * FROM events
WHERE event_type = 'purchase' AND amount > 50

-- String operations
SELECT * FROM events
WHERE user_id LIKE 'user_%'

-- Null checks
SELECT * FROM events
WHERE description IS NOT NULL

UNNEST 运算符

UNNEST 运算符将数组转换为多行。这对于处理列表数据类型很有用。

UNNEST 限制：

只能出现在 SELECT 子句中
每个 SELECT 语句只能展开一个数组

示例：

SELECT
    UNNEST([1, 2, 3]) as numbers
FROM events;

这将产生：

+---------+
| numbers |
+---------+
|       1 |
|       2 |
|       3 |
+---------+

多条语句

Pipeline 可包含多条 INSERT 语句，以分号分隔。每条语句从 stream 读取并写入 sink。使用多条语句可根据内容将单个 stream 中的事件路由到多个 sink。

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

INSERT INTO signups_sink
SELECT user_id, created_at FROM events
WHERE event_type = 'signup';

要在 Wrangler CLI 中提供多条语句，请使用 --sql-file 标志传递文件：

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

有关完整示例，请参阅将 stream 扇出到多个 Iceberg 表。