> ## Documentation Index
> Fetch the complete documentation index at: https://docs.molesignal.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Scheduled pipelines

> Read a source stream on a schedule, apply a VRL transform chain, and write the result to a target stream or external connector.

A **scheduled pipeline** reads a source stream over a time window, applies an ordered chain of VRL
transforms, and writes the output to a target stream (standard intake) — optionally fanning the same
events out to external **connectors** (S3, Kafka). Pipelines run on a simplified interval schedule,
or on demand as a **backfill** over a historical window.

<Note>
  Scheduled pipelines transform data **after intake**, as a recurring job. To transform events on
  the intake hot path instead, attach a [function](/en-US/functions) to
  the intake step.
</Note>

<Frame caption="Scheduled pipeline">
  <img src="https://mintcdn.com/molesignal/W03b-Z-TATDejvIA/images/architecture/pipeline_en_light.svg?fit=max&auto=format&n=W03b-Z-TATDejvIA&q=85&s=8f79b8ddd012387d4796647e033449e0" alt="Scheduled pipeline" className="block dark:hidden" width="528" height="570" data-path="images/architecture/pipeline_en_light.svg" />

  <img src="https://mintcdn.com/molesignal/W03b-Z-TATDejvIA/images/architecture/pipeline_en_dark.svg?fit=max&auto=format&n=W03b-Z-TATDejvIA&q=85&s=16b8847fdcfea1e62ffbaf440e0d9e82" alt="Scheduled pipeline" className="hidden dark:block" width="528" height="570" data-path="images/architecture/pipeline_en_dark.svg" />
</Frame>

## Build a pipeline in the graph editor

Open **Pipelines → New** to lay out the flow on a canvas: **source → transform → sink**. Nodes are
draggable, handles support connections, and the editor validates the graph during layout
(missing source/sink, a transform without a name or script, duplicate stream names, a source that
also appears as a sink). The first source and first stream sink are saved as the pipeline's
`source_stream` and `target_stream`.

| Node          | Meaning                                                     |
| ------------- | ----------------------------------------------------------- |
| **Source**    | The stream the pipeline reads from each run.                |
| **Transform** | One VRL step. Steps run in order, top to bottom.            |
| **Sink**      | A target stream (standard intake) or an external connector. |

## Transforms

Each transform is a [VRL](/en-US/functions) script. The transform inspector supports reuse of a
**built-in preset** or any saved [function](/en-US/functions) from the **reuse** picker. Selecting
an entry fills the step script, which remains editable inline.

Built-in presets ship with every instance and are read-only:

| Preset             | Effect                                                                              |
| ------------------ | ----------------------------------------------------------------------------------- |
| `normalize-logs`   | Parse the JSON `message`, add environment/cluster fields, lowercase `level`.        |
| `route-by-service` | Normalize `service` and derive a per-service target stream name (`logs_<service>`). |
| `parse-key-value`  | Parse a `logfmt` / `key=value` message and merge the fields into the event.         |
| `redact-email`     | Replace email addresses in `message` with a placeholder.                            |
| `add-intake-time`  | Stamp an intake timestamp to help debug end-to-end latency.                         |

<Tip>
  Presets and saved functions share one catalog, with both types on the **Functions** page. Built-in
  presets are badged and read-only; use **Copy script** to create a custom function.
</Tip>

### Extend tables

An **extend table** is a key → record lookup table available to a transform. Open **Extend tables**,
create a table, add rows (a key plus named fields), and perform the lookup from VRL:

```ruby theme={null}
# enrich each event with the service's owning team
.team = lookup("service_meta", .service).team
```

Lookups happen in memory and add no query cost to the run.

## Sinks and connector egress

A sink is either a **target stream** (the default — events are written through standard intake and
become queryable) or an external **connector**:

| Connector | Payload                                                         |
| --------- | --------------------------------------------------------------- |
| **S3**    | Batched JSON Lines `PUT` to `s3://<bucket>/<prefix><id>.jsonl`. |
| **Kafka** | One JSON record produced per event to the configured topic.     |

Add connectors under **Pipelines → Connectors**, then select connectors as sinks in the graph editor. A
pipeline can write to a stream and egress to connectors in the same run.

### Hide the source stream from queries

When fanning a catch-all source stream out into per-service or per-tenant target streams — for
example with the `route-by-service` preset above — the source stream keeps accumulating the raw,
unrouted events. To keep queries and dashboards pointed only at the routed streams, mark the source
stream **non-queryable**: open **Streams → *source* → Settings** and turn **Queryable** off.

A non-queryable stream still accepts and retains data and feeds the pipeline, but the stream is
hidden from the query stream selectors and rejected by SQL and PromQL search with `stream is not
queryable`. Flip the toggle back on at any time.

## Schedule and lookback

| Field           | Notes                                                      |
| --------------- | ---------------------------------------------------------- |
| `cron`          | Interval shorthand: `every:30s`, `every:5m`, `every:1h`.   |
| `lookback_secs` | How far back each run reads from the source (default 300). |
| `enabled`       | Toggle the schedule without deleting the pipeline.         |

The runner does not parse standard cron expressions. A schedule outside the supported `every:`
syntax is skipped.

On each tick the runner reads `[now - lookback_secs, now]` from the source, applies the chain, and
writes the output. Every run is recorded — view history under the pipeline's **Runs** tab, or via
`GET /api/v1/scheduled_pipelines/{id}/runs` (`state`, `scanned_rows`, `error`).

<Note>
  The schedule runner is a singleton and runs only on the **alert-manager**
  (or **standalone**) node, so enabled pipelines fire once per interval rather than once per node.
</Note>

## Backfill

To process a historical window on demand, submit a backfill:

```bash theme={null}
curl -X POST http://localhost:5080/api/v1/scheduled_pipelines/$PIPELINE_ID/backfill \
  -H "authorization: Bearer $MS_JWT" \
  -H 'content-type: application/json' \
  -d '{"start_micros": 1717200000000000, "end_micros": 1717286400000000}'
```

The window must be **≤ 31 days**. The request returns `202` with a `job_id` and a monitor URL; the
backfill runs through the async [search-job](/en-US/distributed-deployment#async-search-job-pipeline)
worker — read the source window, apply the same transform chain, write the target stream and egress.

## Permissions

* `pipelines.read` lists pipelines and reads pipeline run history.
* `pipelines.create` creates a pipeline.
* `pipelines.edit` changes the pipeline graph, transforms, schedule, or lookback.
* `pipelines.pause` enables or pauses scheduled execution.
* `pipelines.run` submits a backfill.
* `pipelines.delete` deletes a pipeline.

<Card title="Pipeline API" icon="code" href="/en-US/api/intake/scheduled-pipelines">
  Create, update, list, and backfill scheduled pipelines over the HTTP API.
</Card>