Manage pipelines

Learn how to:

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

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

To create a pipeline, run the basin pipelines create command:

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"

You can also provide SQL from a file:

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

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

npx wrangler basin pipelines setup

yarn wrangler basin pipelines setup

pnpm wrangler basin pipelines setup

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

Common patterns include:

Transfer all data from stream to sink:

INSERT INTO my_sink SELECT * FROM my_stream

Filter events based on conditions:

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

Choose only the fields you need:

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

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

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.

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

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>

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

npx wrangler basin pipelines list

yarn wrangler basin pipelines list

pnpm wrangler basin pipelines list

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

To delete a pipeline, run the basin pipelines delete command:

npx wrangler basin pipelines delete <PIPELINE_ID>

yarn wrangler basin pipelines delete <PIPELINE_ID>

pnpm wrangler basin pipelines delete <PIPELINE_ID>

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