Pipelines

Chain multiple composable capabilities into one ordered, synchronous, deterministic request. Clean, enrich, analyze, and route a single record set without making a round trip per step.

Endpoint

POST /api/v1/pipeline

Send a data.records array and an ordered steps array. Each step names an operation and optional params. The entire pipeline is validated before any step runs, so an invalid chain fails fast with 400 and nothing is metered.

How steps compose

Every composable operation belongs to a behavior class that defines how it affects the records flowing through the pipeline:

• Transforming (normalize, standardize, transform, redact, mask, map) — rewrite the working records in place and pass them to the next step.
• Enriching (classify, tag) — add fields to each record while preserving the record count.
• Reducing (dedupe) — may drop records; downstream steps see the smaller set.
• Analysis (validate, data-quality, completeness, consistency, anomalies, conflicts) — attach a report under artifacts and pass records through unchanged.
• Decision (route) — a terminal step that assigns each record a destination. It must be the last step.

Composition rules

• Steps run top to bottom; each step's output is the next step's input.
• A terminal step (route) may only appear as the final step — nothing can consume its output.
• Pairwise operations (dedupe, conflicts) accept a bounded number of records inside a pipeline to keep runs fast and predictable.
• Operations not listed here (for example two-source reconcile or the scoring engines) are not composable and are rejected during validation.

Example: clean, redact, classify

POST /api/v1/pipeline
Authorization: Bearer zap_live_...
Content-Type: application/json

{
  "data": {
    "records": [
      { "company": " ACME, INC. ", "email": "sales@acme.com", "note": "urgent renewal" }
    ]
  },
  "steps": [
    { "operation": "normalize" },
    { "operation": "redact", "params": { "types": ["email"] } },
    { "operation": "classify", "params": {
      "field": "note",
      "categories": [{ "name": "renewal", "keywords": ["renewal"] }]
    } }
  ]
}

{
  "request_id": "req_9f2c1a7b3e5d4c8f0a1b2c3d",
  "records": [
    { "company": "ACME, Inc.", "email": "[redacted]", "note": "urgent renewal", "category": "renewal" }
  ],
  "steps": [
    { "operation": "normalize", "behavior": "transforming", "status": "success", "records_in": 1, "records_out": 1, "billable_units": 1, "processing_ms": 0 },
    { "operation": "redact", "behavior": "transforming", "status": "success", "records_in": 1, "records_out": 1, "billable_units": 1, "processing_ms": 0 },
    { "operation": "classify", "behavior": "enriching", "status": "success", "records_in": 1, "records_out": 1, "billable_units": 1, "processing_ms": 0 }
  ],
  "artifacts": {},
  "records_processed": 3,
  "usage": { "credits_used": 3, "credits_remaining": 997 }
}

The response returns the final records, a per-step steps report (behavior, record counts, and timing), and any analysis artifacts.

Example: analyze, then route

Analysis steps do not change records — they attach a report you can read from artifacts while later steps keep working on the original data. Here a data-quality report is attached, then records are routed by a rule:

{
  "data": { "records": [ { "id": "1", "email": "a@b.com" }, { "id": "2", "email": "" } ] },
  "steps": [
    { "operation": "data-quality" },
    { "operation": "route", "params": {
      "rules": [{ "name": "missing_email", "when": [{ "field": "email", "op": "missing" }], "then": { "destination": "review" } }],
      "fallback": "ok"
    } }
  ]
}

Billing

Pipeline usage is based on the billable work performed by each capability. Every step reports billable_units — the records it actually processed — and the pipeline is charged the sum of those units, returned as records_processed. If an earlier step reduces the record set, later steps are metered against the records they actually process, so you are never billed for records a step never saw.

Worked example. Send 1,000 records through normalize → dedupe → validate. Normalize processes 1,000; dedupe processes 1,000 and collapses the set to 500; validate then processes 500. Total billable work = 1,000 + 1,000 + 500 = 2,500 units — not 1,000 × 3 = 3,000. With no reducing step the two are identical.

If a step fails, the request aborts with a structured error identifying the step, and nothing is metered — you are never charged for a partial or failed pipeline, and a retry of a failed call cannot double-charge.

Saved pipelines

Prefer to store a pipeline and run it by id with variable substitution? See saved pipelines under POST /api/v1/pipelines/{id}/run, which reuse the same engine, behavior classes, and billing rules.