Write a flow

You will write acme/sample: a data flow that keeps a repeatable sample of a JSONL file. Then you will wrap it so it can run, and try it in the sandbox. This guide assumes the acme pack from Write a pipeline is registered.

1. Pick the stages

A data flow is an ordered list of stages. Each stage is a kernel that reads the working file and writes the next one. These ship with cairn:

Stage	Does
ingest.normalize	reads the source (JSONL, CSV, Parquet) into the working file
ingest.sql, ingest.api	read rows from a SQL source or a paginated JSON API
dedup.exact, dedup.near	drop exact or near duplicates
decontam.apply, decontam.scan	drop, or only report, rows that overlap an evaluation set
pii.redact	redact secrets and personal data
expectations.check	evaluate a declared set of expectations
filter.rows, sample.rows, split	keep matching rows, a sample, or a train/validation/test split
profile.tokens	token-length profile
export.jsonl, export.parquet	write the final file: the flow’s product

A stage that writes declares itself external_write; the flow’s risk is its highest stage risk, and that is what a gate checks.

2. Write the flow

Create acme/flows/sample.yaml:

name: sample
kind: data
description: Keep a repeatable sample of a JSONL file and export it.

stages:
  - stage: ingest.normalize
    config:
      path: $config.path
  - stage: sample.rows
    config:
      n: $config.n
      method: hash
  - stage: export.jsonl

• name is the flow’s address inside the pack: acme/sample.
• $config.path and $config.n are placeholders. Whoever runs the flow fills them. A placeholder nobody fills is dropped, and the stage uses its default.
• method: hash makes the sample depend only on the rows and the seed, so the same input always gives the same sample.

cairn flow ls should now list it:

  acme/sample  kind=data stages=3

3. Wrap it in a pipeline

A flow never runs on its own. Give it a one-leg pipeline, acme/pipelines/take-sample.yaml:

name: take-sample
kind: pipeline
description: Keep a repeatable sample of a JSONL file.

inputs:
  $inherit: take
  path: {type: Dataset, description: "JSONL file to sample."}
  n: {type: int, default: 2, description: "Rows to keep."}

steps:
  - id: take
    flow: sample
    produces: {export: Dataset}

$inherit: take makes every placeholder of the take leg’s flow a pipeline input and wires it through, so the leg needs no with: block. Declaring path and n next to it gives them a type, a default and a description.

$ cairn pipeline show acme/take-sample
acme/take-sample  —  Keep a repeatable sample of a JSONL file.
  hash: sha256:2f48a1eab540fec7
  inputs: n, path

  •  take [flow] acme/sample

4. Try it in the sandbox

cairn remote launch acme/take-sample -i path=/data/chats.jsonl --environment sandbox --watch

Then read what each stage reported:

$ cairn remote findings pl_0bb24fc588284ad1
pipeline pl_0bb24fc588284ad1  status=done
  take  [done] run=run-flow-acme--sample-79560b
    ingest_normalize_rows             4
    sample_rows_method                hash
    sample_rows_n                     2
    sample_rows_rows                  2
    sample_rows_sampled               True
    sample_rows_total                 4

Every stage’s results appear as <stage>_<field>, with the dot in the stage name turned into _. These are the names a gate’s predicate: reads under steps.<leg>.findings and its show: list reads as <leg>.<field>.

ML flows and agentic flows

An ML flow has the same shape with kind: ml. Its stages train or evaluate (for example train.qlora, eval.checkpoints), and it runs on the ML engine the environment configures.

An agentic flow is a markdown file, flows/<name>.md, run by a template: leg. You can iterate on one without a pipeline using the dev loop, cairn run <pack>/<flow> --input in.json --dry-run (or --mock, --interactive); see Flows.

Next

• Write a pipeline: put flows and gates together.
• Pipeline and flow YAML: every flow and stage field.
• Capabilities and environments: where the work runs.