SELECT ステートメント

SELECT ステートメントは、Cloudflare Pipelines でデータを変換するために使います。一般的な形式は次のとおりです。

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

パイプラインは、1 つ以上の INSERT INTO sink SELECT ... FROM stream ステートメントを実行します。同じパイプラインから複数の 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 句は、クエリのデータソースを指定します。テーブル名またはサブクエリです。テーブル名は、ストリーム名、または 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 句にのみ記述できます
1 つの SELECT ステートメントで展開できる配列は 1 つだけです

例:

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

結果は次のとおりです。

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

複数のステートメント

パイプラインには、セミコロンで区切った複数の INSERT ステートメントを含められます。各ステートメントはストリームから読み取り、Sink に書き込みます。1 つのストリームのイベントを内容に応じて複数の 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

具体的な例は 1 つのストリームを複数の Iceberg テーブルへファンアウトするを参照してください。