Pipelines

A pipeline transforms one or more datasets into another dataset. Use it to clean, filter, or combine data before mapping it to your domain model.

Define a pipeline

Define a step with its inputs, output, and transform, then add it to a pipeline with .then(). Export the pipeline from your project's pipelines/ folder.

This example trims campaign names. rawCampaigns and cleanCampaigns are datasets defined in datasets/campaigns.ts, each with id, name, and status columns.

TSpipelines/campaigns.ts

import { definePipeline, definePipelineStep } from "@sixb/core"
import { rawCampaigns, cleanCampaigns } from "../datasets/campaigns"

const cleanNames = definePipelineStep("clean-campaign-names")
  .inputs({ campaigns: rawCampaigns })
  .output(cleanCampaigns)
  .sql(({ campaigns }) => `
    select id, trim(name) as name, status
    from ${campaigns}
  `)

export const prepareCampaigns = definePipeline("prepare-campaigns")
  .then(cleanNames)

Input names become table references in the SQL query. SQL uses the DuckDB dialect and requires a lake provider with SQL transform support.

Each step replaces its output dataset by default. To append rows instead, use .output(cleanCampaigns, { mode: "append" }). When the output dataset has a primaryKey, a replacement stores only the rows that changed, so projections of it apply only those.

Add more steps with .then(nextStep). They run in order, so each can read the previous step's output. If a step fails, outputs from earlier steps remain available.

A run outlives its worker. After a crash or a deploy it resumes at the first step that did not commit, so a step that committed is not run again, and a step that did not may run its code twice. A run that loses its worker five times ends failed.

Run automatically

Create a schedule to run the pipeline when its source dataset changes.

TSschedules/campaigns.ts

import { defineSchedule, events } from "@sixb/core"
import { rawCampaigns } from "../datasets/campaigns"

export const campaignsUpdated = defineSchedule("campaigns-updated")
  .on(events.dataset(rawCampaigns).updated())

Import the schedule and attach it to the pipeline with .when(). If the update's event is lost, the orchestrator notices within two minutes that the pipeline has not read the new version and runs it. A failed run counts as having read what it read, so it is not retried; a pipeline that never ran waits for you or the next update.

// pipelines/campaigns.ts
import { campaignsUpdated } from "../schedules/campaigns"

export const prepareCampaigns = definePipeline("prepare-campaigns")
  .when(campaignsUpdated)
  .then(cleanNames)

TypeScript steps

Use .run() instead of .sql() when a transform needs your own code or a library. Here is the same step written in TypeScript:

import { definePipelineStep } from "@sixb/core"
import { rawCampaigns, cleanCampaigns } from "../datasets/campaigns"

const cleanNames = definePipelineStep("clean-campaign-names")
  .inputs({ campaigns: rawCampaigns })
  .output(cleanCampaigns)
  .run(async ({ inputs, output }) => {
    async function* rows() {
      for await (const row of inputs.campaigns.readRows()) {
        yield { ...row, name: String(row.name).trim() }
      }
    }

    await output.writeRows(rows())
  })

Read only some columns

Pass dataset.pick(...) as an input to read only those columns:

const titles = definePipelineStep("campaign-titles")
  .inputs({ campaigns: rawCampaigns.pick("id", "name") })
  .output(campaignTitles)
  .sql(({ campaigns }) => `select id, upper(name) as name from ${campaigns}`)

A step's output inherits the markings of the columns it reads. A .run() step inherits every one of them, so picking only what it needs keeps sensitive columns out of its output. The output of a step is always the whole dataset.

Next steps