1# Pipelines Configuration
2 
3Templates for creating streams, sinks, and pipelines via CLI, REST, or Terraform. For the full flag/field list and allowed values, pull `https://developers.cloudflare.com/pipelines/reference/wrangler-commands/` and the streams/sinks/pipelines docs.
4 
5## Naming Rules
6 
7- **Streams, sinks, pipelines** use underscores: `my_stream`, `my_sink`, `my_pipeline`.
8- **Buckets** use hyphens: `my-bucket`.
9 
10## Schema (Structured Streams)
11 
12Schema is a JSON object with a `fields` array; each field has `name`, `type`, `required`.
13 
14```json
15{
16  "fields": [
17    { "name": "event_id", "type": "string", "required": true },
18    { "name": "amount", "type": "float64", "required": false }
19  ]
20}
21```
22 
23Field types include `string`, `bool`, `int32/64`, `float32/64`, `timestamp`, `json`, `binary`, `list`, `struct` (with nested `items`/`fields`). For the authoritative type list, see `https://developers.cloudflare.com/pipelines/sql-reference/sql-data-types/`.
24 
25Unstructured streams (no schema) store everything in a single `value` column.
26 
27> Pipelines auto-adds `__ingest_ts` (TIMESTAMP, day-partitioned). Do **not** include it in your schema.
28 
29## Option A: Interactive (Simplest)
30 
31```bash
32npx wrangler pipelines setup   # creates stream + sink + pipeline, optionally bucket + catalog
33```
34 
35## Option B: Wrangler CLI (Explicit)
36 
37```bash
38# 1. Stream
39npx wrangler pipelines streams create my_stream --schema-file schema.json
40 
41# 2. Sink — R2 Data Catalog (Iceberg). Creates the namespace + table.
42npx wrangler pipelines sinks create my_sink \
43  --type r2-data-catalog \
44  --bucket my-bucket --namespace my_namespace --table my_table \
45  --catalog-token $API_TOKEN \
46  --compression zstd --roll-interval 300
47 
48# 2b. Sink — R2 raw Parquet (alternative)
49npx wrangler pipelines sinks create my_sink \
50  --type r2 --bucket my-bucket --format parquet \
51  --path analytics/events --partitioning "year=%Y/month=%m/day=%d" \
52  --access-key-id $KEY --secret-access-key $SECRET
53 
54# 3. Pipeline (SQL connects stream → sink)
55npx wrangler pipelines create my_pipeline \
56  --sql "INSERT INTO my_sink SELECT * FROM my_stream"
57```
58 
59Tuning knobs (`--compression`, `--roll-interval`, `--roll-size`, etc.) and their allowed values/defaults change — pull the wrangler-commands and sinks docs rather than hardcoding. Rule of thumb: prod `--roll-interval 300+`, dev `10` (creates many small files).
60 
61> **⚠️ Pipelines are immutable.** SQL, schema, and sink config can't be changed — delete and recreate.
62 
63## Option C: REST API (Programmatic)
64 
65Base: `https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/pipelines/v1`
66 
67```bash
68# Stream
69curl -X POST "$BASE_URL/streams" -H "Authorization: Bearer $API_TOKEN" \
70  -H "Content-Type: application/json" -d '{
71    "name": "my_stream",
72    "http": {"enabled": true, "authentication": false},
73    "schema": {"fields": [{"name": "event_id", "type": "string", "required": true}]}
74  }'
75 
76# Sink — NOTE REST field names differ from CLI flags (see table)
77curl -X POST "$BASE_URL/sinks" -H "Authorization: Bearer $API_TOKEN" \
78  -H "Content-Type: application/json" -d '{
79    "name": "my_sink", "type": "r2_data_catalog",
80    "config": {"bucket": "my-bucket", "namespace": "my_namespace",
81               "table_name": "my_table", "token": "'$API_TOKEN'",
82               "rolling_policy": {"interval_seconds": 300}},
83    "format": {"type": "parquet"}
84  }'
85 
86# Pipeline
87curl -X POST "$BASE_URL/pipelines" -H "Authorization: Bearer $API_TOKEN" \
88  -H "Content-Type: application/json" \
89  -d '{"name": "my_pipeline", "sql": "INSERT INTO my_sink SELECT * FROM my_stream;"}'
90```
91 
92**REST field names ≠ CLI flags** (common failure — not obvious from docs):
93 
94| REST (config body) | CLI flag | Gotcha |
95|--------------------|----------|--------|
96| `"type": "r2_data_catalog"` | `--type r2-data-catalog` | underscores vs hyphens |
97| `"table_name"` | `--table` | different key |
98| `"token"` | `--catalog-token` | different key |
99| `"format": {"type": "parquet"}` | (implied) | required in REST, omitted in CLI |
100 
101## Worker Binding
102 
103```jsonc
104// wrangler.jsonc
105{ "pipelines": [ { "stream": "<STREAM_ID>", "binding": "MY_STREAM" } ] }
106```
107 
108> Binding field is `"stream"` as of June 2026 (was `"pipeline"`, still accepted). Use the **stream ID** (`wrangler pipelines streams list`), not the pipeline ID. Redeploy after adding. Generate typed bindings with `npx wrangler types` → `Pipeline<Cloudflare.MyStreamRecord>` from `cloudflare:pipelines`.
109 
110## Terraform
111 
112Resources: `cloudflare_pipeline_stream`, `cloudflare_pipeline_sink`, `cloudflare_pipeline`. For current attribute schemas pull `https://developers.cloudflare.com/pipelines/reference/terraform/`.
113 
114```hcl
115resource "cloudflare_pipeline_stream" "my_stream" {
116  account_id     = var.cloudflare_account_id
117  name           = "my_stream"
118  format         = { type = "json" }
119  schema         = { fields = [{ name = "value", type = "json", required = true }] }
120  http           = { enabled = true, authentication = false, cors = {} }
121  worker_binding = { enabled = false }
122}
123 
124resource "cloudflare_pipeline_sink" "my_sink" {
125  account_id = var.cloudflare_account_id
126  name       = "my_sink"
127  type       = "r2_data_catalog"
128  format     = { type = "parquet" }
129  schema     = { fields = [] }
130  config     = {
131    account_id = var.cloudflare_account_id
132    bucket     = cloudflare_r2_bucket.pipeline_bucket.name
133    table_name = "my_table"
134    token      = var.catalog_token
135  }
136}
137 
138resource "cloudflare_pipeline" "my_pipeline" {
139  account_id = var.cloudflare_account_id
140  name       = "my_pipeline"
141  sql        = "INSERT INTO ${cloudflare_pipeline_sink.my_sink.name} SELECT * FROM ${cloudflare_pipeline_stream.my_stream.name}"
142}
143```
144 
145## Credentials
146 
147| Type | Permission |
148|------|------------|
149| Catalog token (Iceberg sink) | R2 Storage Admin R&W + R2 Data Catalog R&W |
150| R2 credentials (raw sink) | Object Read & Write |
151| HTTP ingest token | Workers Pipelines Send (only if stream auth enabled) |
152 
153## See Also
154 
155- [api.md](api.md) — sending events, REST API, lifecycle · [gotchas.md](gotchas.md) — immutability, REST≠CLI
156