> ## Documentation Index > Fetch the complete documentation index at: https://docs.kaireonai.com/llms.txt > Use this file to discover all available pages before exploring further. # Flow Getting Started > 6 steps to build, run, and observe your first IR-native Flow pipeline. 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//edit`. You never paste JSON. ### Or: create a pipeline (API) For automation or CI, the same starter IR can be POSTed: ```bash theme={null} curl -X POST https://playground.kaireonai.com/api/v1/pipelines \ -H "x-api-key: $API_KEY" \ -H "x-tenant-id: $TENANT_ID" \ -H "content-type: application/json" \ -d '{ "name": "orders-ingest", "connectorId": "", "schemaId": "", "irVersion": "1.0", "ir": { "kind": "pipeline", "version": "1.0", "id": "orders-ingest", "metadata": { "name": "orders-ingest" }, "nodes": [ { "id": "src", "kind": "source", "connector": "local_fs", "config": { "path": "/data/orders/", "pattern": { "type": "glob", "value": "*.csv" }, "ordering": "lexicographic", "waitPolicy": { "maxRetries": 3, "intervalMinutes": 5, "onMissAction": "skip" }, "atomicity": { "stagingFolder": ".processing/", "successFolder": ".archive/", "failureFolder": ".failed/" }, "format": "csv" } }, { "id": "tgt", "kind": "target", "input": "src", "schema": "public.ds_orders", "loadMode": "append" } ], "errorHandling": { "dlq": { "enabled": false }, "retry": { "maxAttempts": 1, "backoff": "exponential" } } } }' ``` ## 2. Tour the editor After **Create & Open Editor**, you land on `/data/flow-pipelines//edit`. The shell has two resizable panes plus a top bar and a bottom strip: | Pane | What it shows | | ---------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | **Top bar** | `← Pipelines` link, pipeline name, status pill (`draft` until first successful run), IR version pill (`v1`, `v2`, …), the published-version pill, and **Save changes / Versions / Publish / Run now** buttons | | **Center pane** | Tabs: **Visual / JSON IR / SQL Preview / Lineage / Schedule** | | **Right pane** | **Node Config** — click any node on the Visual canvas to inspect its config; if no node is selected, a tip points you at the docked AI panel | | **Bottom strip** | Last run timestamp + status, next-fire summary, DLQ count, link to docs | 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: ```json theme={null} { "id": "norm", "kind": "transform", "input": "src", "ops": [ { "type": "cast", "field": "amount", "to": "numeric" }, { "type": "rename", "from": "cust_id", "to": "customer_id" }, { "type": "hash", "field": "ssn", "algorithm": "sha256" } ] } ``` 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](/data/transforms/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: ```json theme={null} { "id": "v", "kind": "validate", "input": "norm", "datasetLevel": { "rowCount": { "min": 1, "onFail": "abort" } }, "rowLevel": [ { "type": "range", "field": "amount", "min": 0, "onFail": "warn" } ], "quarantine": { "enabled": true, "table": "dlq.orders" } } ``` **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](/data/pipelines/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 * Master spec for the IR shape: [Pipeline IR](/data/transforms/pipeline-ir) * AI authoring of IR changes: [AI Pipeline Mode](/ai-ml/ai-pipeline-authoring) * External agent surface: [MCP Flow Server](/ai-ml/mcp-flow-server) * File ingestion details: [File Ingestion](/data/connectors/file-ingestion)