Pipeline library

Bulk data movement: read a source, reshape the records, write a sink. Bound by Datapipes, and by AI Flow in its streaming forms.

kis datapipes -f pipeline.yaml -v batch=2026-06 -e env.yaml

This library stands alone. A datapipe binds it and nothing else — no shell:, no ssh:, no git:. See Flow engines for the reasoning; the short version is that a pipeline streams records, and the common atoms are units of work rather than stages in a stream.

Readers

Source	Datapipe task	Script functions
CSV	`read-csv:`	`csv.read`
Excel	`read-excel:`	`excel.read`
Parquet	`read-parquet:`	`parquet.read`
Database	`read-db:`	`db.query`
REST	`read-rest:`, `read-rest-json:`, `read-rest-json-pipe:`	`rest.request`
Standard output	`read-stdout:`	`stdout.exec`
Entity	`readentity:`	`entity.query`
Entity in a datastore	`readentity-in-datastore:`	`entity.query`

Three REST readers exist because three shapes of API do. read-rest: is the plain one; read-rest-json: unwraps a JSON body; read-rest-json-pipe: streams rather than materialising, which is the one to use when the response is larger than you want in memory.

Writers

Sink	Datapipe task	Script functions
CSV	`writecsv:`	`csv.write`
Excel	`writeexcel:`	`excel.write`
Parquet	`writeparquet:`	`parquet.write`
Database	`writedb:`	`db.execute`
REST	`writerest:`	`rest.request`
Entity in a datastore	`writeentity-in-datastore:`	`entity.mutate`

Transforms

Operation	Datapipe task	Script functions
Reshape records	`morph:`	`morph.execute`
Aggregate	`aggregate:`, `aggsum:`	`aggregate.compute`
Sum a numeric column	`sum:`	`compute.sum`
Concatenate	`sumstr:`	`compute.concat`
Duplicate a stream	`replicate:`	`compute.replicate`
Run a script per record	`plugin:`	(you are already in one)
Print records	`print:`	`log.info`
Do nothing	`noop:`	—

morph: is the general transform and where most pipeline logic lives — renaming fields, changing types, dropping rows, deriving values. It runs a script per record: language: picks the runtime, execute: carries the body, and the record arrives as the function’s argument. plugin: is the escape hatch when a record needs real logic, and it runs the same script engine as everything else.

noop: is not filler. It gives a pipeline a named join point or a placeholder stage while the real one is being written, without changing the shape of the graph.

Reading a pipeline

The three parts read in order — source, transform, sink:

name: monthly-export
pipeline: true
tasks:
  - name: source
    read-db:
      query: "select id, name, total from orders where month = '{{batch}}'"
    next:
      go: reshape

  - name: reshape
    morph:
      language: javascript
      execute: |
        function main(record) {
          return {
            id: record.id,
            customer: record.name,
            amount_cents: record.total * 100,
          };
        }
    next:
      go: sink

  - name: sink
    writeparquet:
      path: "./exports/orders-{{batch}}.parquet"

Note pipeline: true — it selects the streaming execution model. Without it the same file is read as an ordinary flow, and the pipeline tasks are not available.

When to use a script instead

A script reaches all of these namespaces and the common ones, so it can read a CSV, shell out and write to S3 in one function. That flexibility is real, and it is also the reason to keep it for one-off work.

For anything recurring, the pipeline is the better home: it streams rather than loading, it records what completed, and the stages are legible to whoever inherits it. Reach for a script when the job is a one-off reshape, and for a pipeline when it will run again.