Pipelines as code

A pipeline is a YAML document that you can version in git. Use openetl plan to review changes and stream delivery guarantees. Use openetl apply to create or update the resources. This guide explains the workflow, resource documents and common pipeline settings.

If you haven't yet, run the quickstart first: this page builds on its faker source and local-postgres destination. Concepts explains the words used here.

Your first pipeline then copies tables from your own Postgres database into another, step by step.

The documentsCopy Anchor Link

Everything you apply is a document with the same four parts:

apiVersion: openetl.io/v1alpha1
kind: Destination
metadata:
  name: local-postgres
spec:
  connector: postgres
  config:
    dsn: { secret: local-postgres-dsn }

• apiVersion is openetl.io/v1alpha1, except for a Connector's, which is openetl.io/v1alpha2.
• kind says what the document defines.
• metadata.name is lowercase letters, digits and hyphens, at most 63 characters. metadata.workspace is optional: an installation has one workspace, default.
• spec is the definition itself.

A file can hold several documents, separated by ---, and -f can be given more than once. Documents refer to each other by name, so a pipeline's source.ref is the name of a Source.

A Source or Destination specifies a connector and its config. The connector's documentation defines the configuration fields. openetl connector spec TARGET prints the connector's specification, including its configuration JSON Schema.

An unversioned name such as postgres can select a newer installed version each time you apply the connection. The plan warns about this. postgres@1 restricts selection to major version 1. Pin an exact version or digest when the deployment requires a specific release.

Choose and manage connectors covers installing official, community and private packages, selecting versions, and upgrading them independently of OpenETL.

SecretsCopy Anchor Link

A config never holds a password. It names a secret instead, as { secret: NAME }:

config:
  dsn: { secret: local-postgres-dsn }

Store secrets with the CLI. set reads a value from standard input or --from-file. With --ref, it stores a reference to a cloud secret store. Runners resolve that reference with their cloud identity:

echo postgres://postgres:quickstart@quickstart-postgres:5432/postgres | openetl secret set local-postgres-dsn
openetl secret list
openetl secret delete local-postgres-dsn

list shows current resources that use each secret. delete refuses a secret used by a current source, destination, webhook, Spark cluster, Flink cluster or pipeline. Check private connector sources separately. Manage connection secrets covers cloud references, credential replacement and deletion.

set drops one line ending from the end of a value it reads, and runners do the same with a value in a cloud store that's text, so a file saved by an editor or written by echo works as is. A field of a JSON value is used exactly as it is.

Plan, then applyCopy Anchor Link

openetl plan -f FILE sends the documents to the API, which checks them and works out what applying them would do, without changing anything. Here's a second pipeline from the quickstart's faker source, into Parquet files on the runner's disk with the built-in parquet connector, at 06:00 every day:

apiVersion: openetl.io/v1alpha1
kind: Destination
metadata:
  name: local-parquet
spec:
  connector: parquet
  config:
    url: file:///tmp/openetl-parquet
---
apiVersion: openetl.io/v1alpha1
kind: Pipeline
metadata:
  name: faker-to-parquet
spec:
  schedule: { cron: "0 6 * * *" }
  source: { ref: faker }
  destination: { ref: local-parquet }

openetl plan -f faker-to-parquet.yaml

would be created workspaces/default/destinations/local-parquet (version d10c3e1925e9)
would be created workspaces/default/pipelines/faker-to-parquet (version known after apply)

workspaces/default/pipelines/faker-to-parquet (version known after apply)
  job size: s (1 vCPU, 2 GiB), the default, in pool default
  streams:
    users      append  exactly_once  exact replay
    purchases  append  exactly_once  exact replay

The plan has two parts:

• A line per document shows its planned state and version. Pipelines affected by a source or destination change also get new versions, even if the file does not define them. Their lines end with "which uses what changed".
• A section per pipeline, which can have:
  • changes: each field that changes, as field: before -> after.
  • job size: how much CPU and memory its jobs get, what set that, and the pool of runners it runs in.
  • streams: each stream's write mode, its delivery guarantee, and the reason for that guarantee.
  • problems: what stops the pipeline from running. plan exits with an error, and apply refuses the whole file.
  • warnings: what's allowed but worth a second look.
  • notes: what the plan couldn't check, and why.

The streams section requires two sources of information:

• The source catalog. openetl apply and openetl discover NAME refresh it. Until discovery completes, the plan checks only delivery settings.
• Destination capabilities. The connector package declares them. For a local executable without a package, the plan explains that the runner checks guarantees before each sync.

A plan with problems looks like this one, for the same pipeline made continuous:

workspaces/default/pipelines/faker-to-parquet (version known after apply)
  job size: s (1 vCPU, 2 GiB), the default, in pool default
  streams:
    users      append  exactly_once  exact replay
    purchases  append  exactly_once  exact replay
  problems:
    stream "users" is read with full_refresh, whose reads end, and a continuous pipeline needs streams read with cdc or stream
    stream "purchases" is read with full_refresh, whose reads end, and a continuous pipeline needs streams read with cdc or stream
openetl: the plan has problems, so apply would refuse it

When the plan is what you want, apply the same files:

openetl apply -f faker-to-parquet.yaml

An apply stores all documents together or stores none. Its output identifies each resource as "created", "updated" or "unchanged", followed by warnings.

After the apply, a runner checks each source and destination that changed or that a changed pipeline uses. The runner also discovers those sources. The output shows checked with their names. A failed check only gives a warning: the applied documents stay, and a new pipeline still starts. To skip the checks, use --no-check. To run the checks for a pipeline at any time, use openetl check pipeline NAME.

A new scheduled pipeline starts its first sync immediately. An updated pipeline waits for its schedule or a manual openetl sync.

openetl apply --dry-run performs apply validation without storing changes. It does not print the plan; use openetl plan to review the planned changes.

VersionsCopy Anchor Link

Each applied document is stored as an immutable version identified by its content hash. A pipeline version pins its source, destination and dbt project versions. Changing a source therefore creates a new version of every pipeline that uses it. Each sync runs one version.

openetl versions faker-to-postgres

VERSION                                                           CURRENT  CREATED              PROVENANCE
edab40da6d76c1aba235c7cd81c5618db8af2206c759539c5baea80a584560b3           2026-11-08 12:30:15  cli:examples/faker-to-postgres.yaml#3
9e052e7e6881672dbdcfcc4356f04f592bdfbd9777622e2d2a877eb1530db134  *        2026-11-08 12:28:23  cli:examples/faker-to-postgres.yaml#3

Provenance identifies the origin of a version. Here, it is the third document in a file applied with the CLI. Versions are identified by content. To restore an earlier pipeline definition, revert its git commit and apply the file. This does not undo changes already written to the destination.

The pipeline spec at a glanceCopy Anchor Link

Here is a pipeline that sets more than the defaults:

apiVersion: openetl.io/v1alpha1
kind: Pipeline
metadata:
  name: shop-to-warehouse
spec:
  schedule: { cron: "CRON_TZ=Europe/Berlin 0 6 * * *", jitter: 5m }
  source:
    ref: shop-db
    streams:
      - name: orders
        sync: incremental
        cursor: updated_at
      - name: customers
        sync: cdc
        columns:
          balance: { type: string }
  destination:
    ref: warehouse
    namespace: shop
    write: merge
    deletes: hard
  schemaPolicy: allow_columns
  delivery:
    require: effectively_once
    commit: { interval: 2m, maxBytes: 512MB, scope: stream }
  resources: { size: m }
  heal: { mode: suggest }

Enum values are lowercase (merge), durations are Go durations (90s, 15m), and sizes in bytes take units (512MB, 64MiB). The tables below give each field in a line; Pipeline in the reference has every field.

RunningCopy Anchor Link

Schedules covers when syncs start: intervals, cron time zones, jitter, a sync that runs long, pausing, and time OpenETL wasn't running.

Source and streamsCopy Anchor Link

DestinationCopy Anchor Link

Schema and deliveryCopy Anchor Link

After and betweenCopy Anchor Link

Integrate covers dbt, Spark and Flink.

Handle schema changes and holds walks through a new column, a refused type change, and repairing and releasing the hold.

Continuous pipelinesCopy Anchor Link

A continuous pipeline has mode: continuous and no schedule. Its job keeps reading and commits at intervals. The default is 10 seconds unless the destination or delivery.commit.interval specifies another value:

apiVersion: openetl.io/v1alpha1
kind: Pipeline
metadata:
  name: orders-live
spec:
  mode: continuous
  source:
    ref: shop-db
    streams:
      - name: orders
        sync: cdc
  destination: { ref: warehouse, namespace: shop }

Its streams must be read in a way that never ends: cdc from a database, or stream from an event stream such as a topic. The plan fails a stream read any other way. A continuous pipeline has no syncs, so it has no postSync steps, and its transform, if any, is Flink. Pausing it stops its job; resuming starts it again from its last checkpoint.

Delivery guarantees in practiceCopy Anchor Link

The default requirement is exactly_once. Set delivery.require to change it for the pipeline, or a stream's require to change it for that stream. OpenETL refuses to run a pipeline if any stream falls below its requirement. The plan lists each stream's guarantee and reason.

Deliver-and-acknowledge destinations can repeat a record after a retry. They provide at-least-once delivery. A pipeline using such a destination must accept that guarantee:

delivery:
  require: at_least_once

Under at_least_once, repeated changes remain as duplicate rows. For example, an incremental read includes rows at the previous highest cursor value again. With a higher guarantee, a capable destination removes duplicates during apply. With at_least_once, it writes them again. Recovery after a failure can also produce duplicates.

When reading a stream delivered at least once, keep one row per _openetl_id. This ID combines the stream, primary key and _openetl_seq. Without a primary key, it uses the row's whole content instead. Repeated reads of the same version produce the same ID. In Postgres:

select distinct on (_openetl_id) *
from faker.users
order by _openetl_id;

For the latest version of each row of an appended stream, keep the highest _openetl_seq for each primary key instead:

select distinct on (id) *
from faker.users
order by id, _openetl_seq desc;

Delivery guarantees explains guarantee selection, plan reasons and streams without primary keys. It also shows how to read data delivered at least once.

Day to dayCopy Anchor Link

Flags go before the names, and every command takes --workspace. openetl COMMAND -h lists a command's flags.

The console shows the same pipelines, syncs, versions and holds; Use the console walks through it.

If a pipeline waits, fails repeatedly or falls behind, start with Troubleshoot a pipeline.