Skip to main content
This walkthrough takes you from zero to a running IR-native pipeline with lineage, error inspection, and a saved schedule. Flow IR is on by default — tenant_settings.flowIrEnabled defaults to true. The flag remains as an explicit kill-switch under /settings if you ever need to disable the pipeline feature wholesale.

​
1. Create a pipeline (UI)

Open Data → Pipelines and click + New Pipeline. The dialog asks for three things:
  • Name — anything human-readable, e.g. orders-ingest.
  • Connector — pick from your existing connectors. Create one under Data → Connectors first if needed.
  • Schema — pick the destination data schema. Create one under Data → Schemas first if needed.
Click Create & Open Editor. The platform builds a starter IR (one source node mapped from your connector’s type, one append-mode target node pointing at your schema) and redirects you straight into the visual editor at /data/flow-pipelines/<id>/edit. You never paste JSON.

​
Or: create a pipeline (API)

For automation or CI, the same starter IR can be POSTed:

​
2. Tour the editor

After Create & Open Editor, you land on /data/flow-pipelines/<id>/edit. The shell has two resizable panes plus a top bar and a bottom strip: Run history is not a tab here — it lives on the standalone Pipeline Runs page (/data/flow-runs). The AI panel is docked at the top-right of the page, not inside the right pane. Click the src node on the Visual canvas — the right pane shows the full source-node IR (path, pattern, ordering, format, atomicity, waitPolicy). Click × in the top-right of the right pane to deselect.

​
3. Add transforms + validation

Two ways to add nodes:

​
a. Visual canvas Add-Node toolbar (one click)

Above the canvas there’s a row of buttons: **+ Transform / + Validate /
  • Enrich / + Branch / + Target**. Click a non-target button — kaireon inserts a default node between the last upstream and your target, repoints the target to consume the new node, and bumps the IR version. The new node is auto-selected so you can immediately edit its config in the right pane or the JSON IR tab. Defaults are intentionally placeholder (e.g. a transform with a single rename: from_field → to_field op) so you know exactly what to fix. (Adding an Enrich node is fine for authoring, but note the enrich executor throws at runtime — remove it before you run the pipeline. Source and Join are not in the toolbar; add those via the JSON IR tab or the AI panel.)

​
b. JSON IR tab (full control)

Add a cast / rename / hash transform. Insert a transform node between source and target, then point tgt.input at the transform:
19 transform op types are accepted by the IR schema. You configure them in the transform node’s op editor (right pane) or by hand in the JSON IR tab. cast, rename, drop, filter, add_field, hash, mask_pii, map_values, split, merge, deduplicate, aggregate, lookup_join, rule-mode sentiment_score, and language_detect run at runtime; expression is limited to LOWER/UPPER/TRIM/LENGTH of a single column. summarize, vector_embed, geo_resolve, and sentiment_score in llm mode are passthrough (rows returned unchanged until a provider is wired). See Transforms for the full reference. Add row-level + dataset-level validation. Row-level rules use a type (notNull / regex / range / fieldType / maxLength), a field, and onFail (abort / warn / skip). Failing rows are written to the DLQ table when quarantine.enabled is set:
Repoint the target + save. Change tgt.input to "v", click Save in the JSON IR tab. parsePipelineIR runs server-side; if the IR is invalid, the structured errors render verbatim on the page. On success the IR version pill bumps (v1 → v2). Or skip the JSON entirely and use the docked AI panel: “Add a sha256 hash on ssn and quarantine rows where amount ≤ 0” produces the same change as a proposal you can accept with one click.

​
4. Run the pipeline

Hit Run now in the top bar. The status badge stays draft until the first run succeeds — then it flips to active.

​
5. Inspect

  • Pipeline Runs page (/data/flow-runs, Sidebar → Data → Pipeline Runs) → your run appears as a row with Status, Started, Completed, Rows In, Rows Out, Processed, and an Error column
  • Lineage tab → pick public.ds_orders → see your loaded rows with their _kaireon_lineage envelopes; click a row to walk back through the IR
  • SQL Preview tab → see the exact INSERT … SELECT the runtime built from your column schema
  • Validation failures → a failing dataset-level check surfaces in the run’s Error column; rows rejected by a row-level rule are written to the DLQ table you named in quarantine.table when quarantine.enabled is set

​
6. Schedule it

Switch to the Schedule tab. Pick cron, click “Daily at 09:00”, choose your timezone, watch the “Next 5 fires” preview, hit Save schedule. The IR gets a schedule field; the in-process scheduler (running every 60 seconds since server startup) picks it up automatically — no external cron, no extra service to deploy. See Flow Scheduler for tuning knobs and how to disable in-process scheduling if you bring your own orchestrator. Click Tick scheduler now in the same tab to fire any due pipelines immediately, useful right after saving a new schedule.

​
Where to go next