Pipeline

Source: src/Cloudflare/Pipelines/Pipeline.ts

A Cloudflare SQL Pipeline — the transform of the Pipelines product. A pipeline is a single SQL statement that reads events from a Stream and writes them to a Sink, both referenced by name.

The SQL is fixed at creation: changing it (or the name) triggers a replacement. Nothing references a pipeline downstream, so replacements are cheap.

Creating a Pipeline

Stream → Sink passthrough

const stream = yield* Cloudflare.Pipelines.Stream("events", {});
const sink = yield* Cloudflare.Pipelines.Sink("events-sink", {
  type: "r2",
  config: { bucket: bucket.bucketName, credentials },
});

const pipeline = yield* Cloudflare.Pipelines.Pipeline("etl", {
  sql: Output.interpolate`INSERT INTO ${sink.name} SELECT * FROM ${stream.name}`,
});

Filtering transform

const pipeline = yield* Cloudflare.Pipelines.Pipeline("errors-only", {
  sql: Output.interpolate`INSERT INTO ${sink.name} SELECT * FROM ${stream.name} WHERE level = 'error'`,
});