Manage pipelines

Learn how to:

Create pipelines with SQL transformations
View pipeline configuration and SQL
Delete pipelines when no longer needed

Create a pipeline

Pipelines execute SQL statements that define how data flows from streams to sinks.

Dashboard

In the Cloudflare dashboard, go to the Pipelines page.
Go to Pipelines
Select Create Pipeline to launch the pipeline creation wizard.
Follow the wizard to configure your stream, sink, and SQL transformation.

Wrangler CLI

To create a pipeline, run the pipelines create command:

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

You can also provide SQL from a file:

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

Alternatively, to use the interactive setup wizard that helps you configure a stream, sink, and pipeline, run the pipelines setup command:

npx wrangler pipelines setup

SQL transformations

Pipelines support SQL statements for data transformation. For complete syntax, supported functions, and data types, see the SQL reference.

Common patterns include:

Basic data flow

Transfer all data from stream to sink:

INSERT INTO my_sink SELECT * FROM my_stream

Filtering events

Filter events based on conditions:

INSERT INTO my_sink
SELECT * FROM my_stream
WHERE event_type = 'purchase' AND amount > 100

Selecting specific fields

Choose only the fields you need:

INSERT INTO my_sink
SELECT user_id, event_type, timestamp, amount
FROM my_stream

Transforming data

Apply transformations to fields:

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

Route one stream to multiple tables

A single pipeline can run multiple INSERT statements, separated by semicolons. Each statement reads from the same stream and writes to a different sink, so you can route ("fan out") events from one stream into several tables based on their content.

This avoids running a separate pipeline for each destination. Each statement filters the stream with its own WHERE clause and projects only the columns relevant to that table. This can also be a used as a cost optimization as you will be billed once for the transformations, not per statement.

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';

For a complete example that fans a live event stream out into five tables, refer to Fan out a stream to multiple Iceberg tables.

View pipeline configuration

Dashboard

In the Cloudflare dashboard, go to the Pipelines page.
Select a pipeline to view its SQL transformation, connected streams/sinks, and associated metrics.

Wrangler CLI

To view a specific pipeline, run the pipelines get command with either the pipeline ID or pipeline name:

npx wrangler pipelines get <PIPELINE_NAME_OR_ID>

To list all pipelines in your account, run the pipelines list command:

npx wrangler pipelines list

Delete a pipeline

Deleting a pipeline stops data flow from the connected stream to sink.

Dashboard

In the Cloudflare dashboard, go to the Pipelines page.
Select the pipeline you want to delete. 3. In the Settings tab, and select Delete.

Wrangler CLI

To delete a pipeline, run the pipelines delete command:

npx wrangler pipelines delete <PIPELINE_ID>

Limitations

Pipeline SQL cannot be modified after creation. To change the SQL transformation, you must delete and recreate the pipeline.