> ## 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. # Data Platform End-to-End Walkthrough > Copy-paste cURL recipe: create an S3 connector with IAM-role auth, define a customer schema with a custom primary key, build a pipeline IR, run it through every load mode, and verify rows landed. Captured from a real 2026-05-02 end-to-end test. This page is the working-example companion to the platform docs. Every cURL block below was actually executed against a live local platform during the 2026-05-02 testing pass — copy, paste, change the IDs, run. ## Setup assumptions * Platform running at `http://localhost:3000` (or substitute your host). * A user is signed in; cookies are persisted in `kc.txt` (see [API Authentication](/api-reference/auth) for getting a session cookie programmatically). For server-to-server, swap the `-b kc.txt` for `-H "X-API-Key: krn_..."` — the endpoints in this walkthrough (connectors, schemas, pipelines) are **control-plane**, so the key must be minted with the `control-plane` scope (a default data-plane-only key gets `403`; see [API Keys](/api-reference/api-keys)). * AWS CLI configured with creds that can `aws sts assume-role` against the role you're about to create. ## 1. Create the IAM role the connector will assume Trust policy lets your local IAM identity assume the role with an external ID; the policy grants minimal S3 access on one bucket. ```bash theme={null} cat > /tmp/trust.json <<'JSON' { "Version": "2012-10-17", "Statement": [{ "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam:::user/" }, "Action": "sts:AssumeRole", "Condition": { "StringEquals": { "sts:ExternalId": "kaireon-test-2026-05-02" } } }] } JSON cat > /tmp/s3-policy.json <<'JSON' { "Version": "2012-10-17", "Statement": [{ "Effect": "Allow", "Action": ["s3:GetObject","s3:ListBucket","s3:PutObject","s3:DeleteObject","s3:CopyObject"], "Resource": ["arn:aws:s3:::","arn:aws:s3:::/*"] }] } JSON aws iam create-role --role-name kaireon-customer-test --assume-role-policy-document file:///tmp/trust.json aws iam put-role-policy --role-name kaireon-customer-test --policy-name s3-access --policy-document file:///tmp/s3-policy.json # IAM is eventually consistent; wait ~15s before first AssumeRole. sleep 15 aws sts assume-role --role-arn arn:aws:iam:::role/kaireon-customer-test \ --role-session-name verify --external-id kaireon-test-2026-05-02 --duration-seconds 900 ``` ## 2. Upload a sample CSV ```bash theme={null} aws s3 cp ./customers.csv s3:///customer-test-2026-05-02/customers.csv ``` The example dataset has the columns `customer_id, first_name, last_name, email, phone, address, city, state, zip_code, country, signup_date, segment, status, lifetime_value, orders_count`. ## 3. Create the S3 connector via the API `POST /api/v1/connectors` with `authMethod: "iam_role"` triggers the STS AssumeRole + HeadBucket + ListObjectsV2 verification path. ```bash theme={null} curl -s -b kc.txt -X POST http://localhost:3000/api/v1/connectors \ -H "X-Requested-With: XMLHttpRequest" -H "Content-Type: application/json" \ -d '{ "name": "Customer S3 (IAM Role)", "type": "aws_s3", "description": "Reads customer.csv via assumed IAM role", "config": { "bucket": "", "region": "us-east-1", "prefix": "customer-test-2026-05-02/" }, "authMethod": "iam_role", "authConfig": { "roleArn": "arn:aws:iam:::role/kaireon-customer-test", "externalId": "kaireon-test-2026-05-02" } }' # → { "id": "", "type": "aws_s3", "authMethod": "iam_role", "status": "active", … } ``` Verify the connection (this exercises STS AssumeRole + HeadBucket end-to-end): ```bash theme={null} curl -s -b kc.txt -X POST http://localhost:3000/api/v1/connectors/test \ -H "X-Requested-With: XMLHttpRequest" -H "Content-Type: application/json" \ -d '{ "id": "" }' # → { ..., "status": "active" } (status flips to "error" with lastError if AssumeRole fails) ``` ## 4. Create the schema with `customer_id` as the primary key When any field is marked `isPrimaryKey: true`, the auto-generated `id BIGSERIAL` column is **skipped** and your column becomes the table's PK. See [Data Model](/data/schemas/data-model). ```bash theme={null} curl -s -b kc.txt -X POST http://localhost:3000/api/v1/schemas \ -H "X-Requested-With: XMLHttpRequest" -H "Content-Type: application/json" \ -d '{ "name": "customer", "displayName": "Customer", "entityType": "customer", "fields": [ { "name": "customer_id", "displayName": "Customer Id", "dataType": "varchar", "isPrimaryKey": true, "isNullable": false, "isUnique": true, "ordinal": 1 }, { "name": "first_name", "displayName": "First Name", "dataType": "varchar", "isNullable": true, "ordinal": 2 }, { "name": "last_name", "displayName": "Last Name", "dataType": "varchar", "isNullable": true, "ordinal": 3 }, { "name": "email", "displayName": "Email", "dataType": "varchar", "isNullable": true, "ordinal": 4 }, { "name": "phone", "displayName": "Phone", "dataType": "varchar", "isNullable": true, "ordinal": 5 }, { "name": "address", "displayName": "Address", "dataType": "varchar", "isNullable": true, "ordinal": 6 }, { "name": "city", "displayName": "City", "dataType": "varchar", "isNullable": true, "ordinal": 7 }, { "name": "state", "displayName": "State", "dataType": "varchar", "isNullable": true, "ordinal": 8 }, { "name": "zip_code", "displayName": "Zip Code", "dataType": "varchar", "isNullable": true, "ordinal": 9 }, { "name": "country", "displayName": "Country", "dataType": "varchar", "isNullable": true, "ordinal": 10 }, { "name": "signup_date", "displayName": "Signup Date", "dataType": "date", "isNullable": true, "ordinal": 11 }, { "name": "segment", "displayName": "Segment", "dataType": "varchar", "isNullable": true, "ordinal": 12 }, { "name": "status", "displayName": "Status", "dataType": "varchar", "isNullable": true, "ordinal": 13 }, { "name": "lifetime_value", "displayName": "Lifetime Value", "dataType": "numeric", "precision": 12, "scale": 2, "isNullable": true, "ordinal": 14 }, { "name": "orders_count", "displayName": "Orders Count", "dataType": "integer", "isNullable": true, "ordinal": 15 } ] }' # → { "id": "", "tableName": "ds_customer", "fields": [ ... 15 ... ] } ``` Inspect the resulting Postgres table to confirm `customer_id` is the PK and there is no auto `id` column: ```bash theme={null} psql "$DATABASE_URL" -c '\d ds_customer' # Indexes: # "ds_customer_pkey" PRIMARY KEY, btree (customer_id) ``` ## 5. Create the pipeline IR The IR has four nodes: `source` → `transform` → `validate` → `target` (upsert keyed on `customer_id`, with the safety bundle: empty-source guard on, validate quarantine to a DLQ table). Note the required `irVersion: "1.0"` and the `errorHandling` block. ```bash theme={null} curl -s -b kc.txt -X POST http://localhost:3000/api/v1/pipelines \ -H "X-Requested-With: XMLHttpRequest" -H "Content-Type: application/json" \ -d '{ "name": "Customer File Test", "description": "S3 → transform → validate → upsert with full safety bundle", "connectorId": "", "schemaId": "", "irVersion": "1.0", "ir": { "kind": "pipeline", "version": "1.0", "id": "customer-file-test", "metadata": { "name": "Customer File Test", "owner": "you" }, "errorHandling": { "dlq": { "enabled": false }, "retry": { "maxAttempts": 3, "backoff": "exponential" } }, "nodes": [ { "id": "src", "kind": "source", "connector": "s3", "config": { "path": "customer-test-2026-05-02/", "bucket": "", "format": "csv", "pattern": { "type": "glob", "value": "*.csv" }, "ordering": "latest_by_mtime", "atomicity": { "stagingFolder": ".processing/", "successFolder": ".archive/{YYYY-MM-DD}/", "failureFolder": ".failed/" }, "waitPolicy": { "maxRetries": 2, "intervalMinutes": 1, "onMissAction": "alert" } } }, { "id": "uppercase_state", "kind": "transform", "input": "src", "ops": [ { "type": "expression", "outputField": "state", "formula": "concat(state)" }, { "type": "rename", "from": "zip_code", "to": "zip_code" } ] }, { "id": "validate_email", "kind": "validate", "input": "uppercase_state", "rowLevel": [ { "type": "regex", "field": "email", "pattern": "^[^@]+@[^@]+\\.[^@]+$", "onFail": "warn" }, { "type": "notNull", "field": "customer_id", "onFail": "abort" } ], "datasetLevel": { "rowCount": { "min": 1, "max": 1000000, "onFail": "abort" } }, "quarantine": { "enabled": true, "table": "dlq.customer_rejected" } }, { "id": "tgt", "kind": "target", "input": "validate_email", "schema": "public.ds_customer", "loadMode": "upsert", "upsertKey": ["customer_id"], "failOnEmptySource": true } ] } }' # → { "id": "", "irVersion": "1.0", "status": "draft", ... } ``` ## 6. Run the pipeline ```bash theme={null} curl -s -b kc.txt -X POST http://localhost:3000/api/v1/pipelines//run \ -H "X-Requested-With: XMLHttpRequest" -H "Content-Type: application/json" -d '{}' ``` Expected response on success: ```json theme={null} { "runId": "9f71b01c-b90d-4dab-8fc0-6e2d05d417a0", "irVersion": 1, "ok": true, "nodeResults": [ { "id": "src", "kind": "source", "rowsOut": 100, "meta": { "matchedFiles": 1, "archivedTo": ".archive/2026-05-02/" } }, { "id": "uppercase_state", "kind": "transform", "rowsOut": 100 }, { "id": "validate_email", "kind": "validate", "rowsOut": 100, "meta": { "rowLevelRules": 2, "rowLevelFailed": 0 } }, { "id": "tgt", "kind": "target", "rowsLoaded": 100, "meta": { "missingFromSource": ["created_at", "updated_at"] } } ] } ``` ## 7. Switch the load mode and re-run The IR is updated via `POST /api/v1/pipelines/:id/ir` (creates a new version atomically). Pattern: GET the current IR, mutate the `target` node, POST the new envelope. Re-upload the source file each run because the source executor archives it after success. ### Switch to `truncate` (with the empty-source guard) ```bash theme={null} # Get current IR curl -s -b kc.txt http://localhost:3000/api/v1/pipelines//ir \ -H "X-Requested-With: XMLHttpRequest" > /tmp/ir.json # Mutate the target node with jq (or your favorite JSON tool) jq '.ir.nodes |= map(if .kind == "target" then .loadMode = "truncate" | del(.upsertKey) else . end) | { ir, comment: "switch to truncate" }' \ /tmp/ir.json > /tmp/ir-truncate.json curl -s -b kc.txt -X POST http://localhost:3000/api/v1/pipelines//ir \ -H "X-Requested-With: XMLHttpRequest" -H "Content-Type: application/json" \ -d @/tmp/ir-truncate.json ``` Re-upload the file and run: ```bash theme={null} aws s3 cp ./customers.csv s3:///customer-test-2026-05-02/customers.csv curl -s -b kc.txt -X POST http://localhost:3000/api/v1/pipelines//run \ -H "X-Requested-With: XMLHttpRequest" -H "Content-Type: application/json" -d '{}' ``` ### Test the empty-source guard Don't re-upload the file. The next run produces an empty source, and the runtime aborts `truncate` BEFORE the destructive statement so your table is preserved: ```bash theme={null} curl -s -b kc.txt -X POST http://localhost:3000/api/v1/pipelines//run \ -H "X-Requested-With: XMLHttpRequest" -H "Content-Type: application/json" -d '{}' # → { "ok": false, "failedNodeId": "tgt", # "error": "target public.ds_customer: source is empty (0 rows from validate_email); # aborting truncate to prevent data loss. Set failOnEmptySource:false to override." } ``` A `warning` System Health alert ("Pipeline source had no files") and an `error` alert ("Pipeline run failed: Customer File Test") fire automatically. Read them via `GET /api/v1/system-health` (see [System Health](/operations-reporting/system-health)). ### Switch to `blue_green` ```bash theme={null} jq '.ir.nodes |= map(if .kind == "target" then .loadMode = "blue_green" | del(.upsertKey, .watermarkColumn) else . end) | { ir, comment: "switch to blue_green" }' \ /tmp/ir.json > /tmp/ir-bg.json curl -s -b kc.txt -X POST http://localhost:3000/api/v1/pipelines//ir \ -H "X-Requested-With: XMLHttpRequest" -H "Content-Type: application/json" -d @/tmp/ir-bg.json ``` Run as before. The runtime loads to `ds_customer_new`, then issues the atomic 3-step rename (target → target\_old, target\_new → target, DROP target\_old). Empty-source guard applies — same protection as truncate. ### Switch to `incremental_watermark` ```bash theme={null} jq '.ir.nodes |= map(if .kind == "target" then .loadMode = "incremental_watermark" | .watermarkColumn = "signup_date" | del(.upsertKey) else . end) | { ir, comment: "switch to watermark" }' \ /tmp/ir.json > /tmp/ir-wm.json curl -s -b kc.txt -X POST http://localhost:3000/api/v1/pipelines//ir \ -H "X-Requested-With: XMLHttpRequest" -H "Content-Type: application/json" -d @/tmp/ir-wm.json ``` The first run reads `MAX(signup_date)` from the target and writes the high-water to `pipeline_watermarks`. Subsequent runs only load rows where `signup_date > `. The INSERT and watermark upsert run inside a single Postgres transaction so a failed load rolls back the watermark advance. ### Try `cdc_mirror` `cdc_mirror` applies the batch as a change feed: an op column on each source row drives upsert vs delete by the CDC key. Set `cdcOpColumn` and `upsertKey` on the target (the target table needs a unique constraint on the key columns): ```bash theme={null} jq '.ir.nodes |= map(if .kind == "target" then .loadMode = "cdc_mirror" | .cdcOpColumn = "op" | .upsertKey = ["customer_id"] | del(.watermarkColumn) else . end) | { ir }' \ /tmp/ir.json > /tmp/ir-cdc.json curl -s -b kc.txt -X POST http://localhost:3000/api/v1/pipelines//ir \ -H "X-Requested-With: XMLHttpRequest" -H "Content-Type: application/json" -d @/tmp/ir-cdc.json ``` Each source row's `op` value (case-insensitive) selects the action: `insert`/`update`/`upsert`/`I`/`U`/`c`/`r` upsert on `customer_id`; `delete`/`D` deletes that key. Any other op value fails the run before any DML — unknown ops are never silently skipped, and the op column is never written to the target. ## 8. Verify rows landed ```bash theme={null} psql "$DATABASE_URL" -c "SELECT count(*), max(signup_date) FROM ds_customer;" ``` For lineage: ```bash theme={null} curl -s -b kc.txt 'http://localhost:3000/api/v1/lineage?schema=public&table=ds_customer&limit=5' \ -H "X-Requested-With: XMLHttpRequest" # → { "rows": [ { "row": { "customer_id": "CUST-00001", "lifetime_value": "10576.95", ... }, # "lineage": { "runId": "...", "pipelineId": "...", "sourceNodeId": "src" } } ], ... } ``` The `lifetime_value` arrives as a string ("10576.95") because the route runs through the `safeJson` helper that converts Postgres `numeric` / Decimal to its string form (preserves precision past `Number.MAX_SAFE_INTEGER`). See [Flow Lineage](/data/observability/flow-lineage). ## 9. Read the System Health feed Every operational error from the run path lands here: ```bash theme={null} curl -s -b kc.txt 'http://localhost:3000/api/v1/system-health?limit=10' \ -H "X-Requested-With: XMLHttpRequest" ``` Mark one read: ```bash theme={null} curl -s -b kc.txt -X PATCH 'http://localhost:3000/api/v1/system-health//read' \ -H "X-Requested-With: XMLHttpRequest" ``` See [System Health](/operations-reporting/system-health) for the full API surface (read-all, dismiss, severity / source filtering, retention purge cron). ## 10. Optional: drive the same pipeline via MCP The MCP server exposes the same operations as tools. From an MCP-aware client (Claude Desktop, Cursor): | Tool | What it does | | ---------------------- | ------------------------------------------------ | | `listFlowPipelines` | Returns the same data as `GET /api/v1/pipelines` | | `runFlowPipeline` | Same as `POST /api/v1/pipelines/:id/run` | | `getFlowPipelineIr` | Same as `GET /api/v1/pipelines/:id/ir` | | `updateFlowPipelineIr` | POST a new IR version | | `inspectFlowError` | Returns DLQ rows for the latest failed run | | `replayFlowRun` | Re-runs a pipeline from a specific run id | Start the server: `cd platform && npm run mcp` (set `KAIREON_API_KEY` and `KAIREON_TENANT_ID` in env). See [MCP Flow Server](/ai-ml/mcp-flow-server). ## Cleanup ```bash theme={null} # Drop the schema (cascades to ds_customer) curl -s -b kc.txt -X DELETE 'http://localhost:3000/api/v1/schemas?id=' \ -H "X-Requested-With: XMLHttpRequest" # Delete the connector curl -s -b kc.txt -X DELETE 'http://localhost:3000/api/v1/connectors?id=' \ -H "X-Requested-With: XMLHttpRequest" # Tear down the IAM role aws iam delete-role-policy --role-name kaireon-customer-test --policy-name s3-access aws iam delete-role --role-name kaireon-customer-test ```