Preprocessor

A preprocessor is an optional REST endpoint that is called before data quality and transformation. It receives the ingested data and can modify, enrich, filter, or replace it before the pipeline continues processing. Use a preprocessor when you need to call out to an external system for data enrichment, format conversion, or custom logic that runs before AI-powered data quality and transformation.

Processing Flow

Source Data Ingested
  |
  v
Preprocessor (optional) <--- you are here
  |
  v
Data Quality
  |
  v
Transformation
  |
  v
Destinations

Configuration

Add a preprocessor section to the pipeline configuration:

{
  "name": "stock_price",
  "source": { ... },
  "preprocessor": {
    "endpoint": "http://my-service:5500/preprocess/sync",
    "async": false,
    "bearerToken": null,
    "timeoutMs": 300000
  },
  "destination": { ... }
}

Fields

Field	Type	Default	Description
`endpoint`	string	Required	URL of the preprocessing service
`async`	boolean	`false`	Synchronous or asynchronous mode
`bearerToken`	string	`null`	Optional `Authorization: Bearer` token
`apiKey`	string	`null`	Optional API key sent to the endpoint as an `x-api-key` header
`timeoutMs`	int	`300000`	Request timeout in milliseconds
`timeoutSeconds`	int	`0`	Legacy timeout in seconds; used only when `timeoutMs` is `0` or omitted. Prefer `timeoutMs`
`batchSize`	int	`0`	Rows per call for row-based (CSV) payloads. `0` makes one call carrying every row; `> 0` sends batches of that many rows, each request also carrying `batch` and `ofBatches`, and stitches every response’s `data.rows` in call order. See Batching

Synchronous Mode

In synchronous mode (async: false), the pipeline POSTs the data to the endpoint and waits for the response. The response replaces the pipeline data (with batchSize > 0, the stitched responses do — see Batching).

Request Payload

{
  "pipelineToken": "5b2f4a1d-8c7e-4f0a-9b3d-6e1c2a4f8b9e",
  "pipelineName": "stock_price",
  "data": {
    "size": 1024,
    "header": ["symbol", "price", "date"],
    "rows": [
      "AAPL,150.25,2026-03-15",
      "GOOG,2800.00,2026-03-15"
    ]
  }
}

For JSON/XML sources, header and rows are omitted and rawData contains the raw content (null fields are left out of the payload rather than sent as null).

Expected Response

Return the (optionally modified) data in the same format:

{
  "data": {
    "size": 1024,
    "header": ["symbol", "price", "date"],
    "rows": [
      "AAPL,150.25,2026-03-15",
      "GOOG,2800.00,2026-03-15"
    ]
  }
}

Any of header, rows, or rawData you leave out of the response keeps its original value. You can modify, add, or remove rows. The pipeline continues with whatever data is returned. If you return an error field, processing is aborted:

{
  "error": "Data validation failed: missing required field"
}

Batching

By default (batchSize = 0) the pipeline makes exactly one call carrying every row, and that call has to fit in memory (it is subject to PIPELINE_MATERIALIZE_MAX_MB). Set batchSize to a positive number to send row-based (CSV) payloads in batches of that many rows, one call per batch, in source order, streamed from disk. Each batched request carries the same envelope plus two extra top-level fields, batch (1-based) and ofBatches, and its data.rows holds only that batch. Every response is read exactly like the single-call response — data.rows replaces that batch’s rows, and a response that leaves rows out keeps them — and the results are stitched, in call order, into the data the pipeline continues with. The header from the first response that carries one applies to the whole payload. JSON/XML payloads and async callbacks are unaffected by batchSize beyond one callback being awaited per batch.

Asynchronous Mode

In asynchronous mode (async: true), the pipeline POSTs the data and then waits for a callback rather than using the response directly. This is useful for long-running preprocessing tasks.

Flow

Pipeline POSTs data to the preprocessor endpoint
Preprocessor returns immediately (e.g., 200 OK)
Preprocessor processes data in the background
Preprocessor POSTs the result back to the pipeline’s callback endpoint
Pipeline resumes with the returned data

Callback Endpoint

The preprocessor sends the result to:

POST http://{pipeline-host}:8080/api/v1/restendpoint/callback

Callback payload:

{
  "pipelineToken": "5b2f4a1d-8c7e-4f0a-9b3d-6e1c2a4f8b9e",
  "pipelineName": "stock_price",
  "data": {
    "size": 1024,
    "header": ["symbol", "price", "date"],
    "rows": ["AAPL,150.25,2026-03-15"]
  }
}

The pipelineToken must match the token from the original request so the pipeline can correlate the callback with the waiting job. If the callback is not received within timeoutMs milliseconds, the pipeline aborts with a timeout error.

Example: Preprocessor Service

A complete working example is provided in examples/preprocessor/app.py. This Python Flask application implements both synchronous and asynchronous preprocessing:

# Synchronous - process and return immediately
@app.route('/preprocess/sync', methods=['POST'])
def preprocess_sync():
    payload = request.get_json()
    data = payload.get('data')

    # Modify data here if needed

    return jsonify({
        'pipelineToken': payload.get('pipelineToken'),
        'pipelineName': payload.get('pipelineName'),
        'data': data
    })

# Asynchronous - return immediately, send callback later
@app.route('/preprocess/async', methods=['POST'])
def preprocess_async():
    payload = request.get_json()
    threading.Thread(
        target=send_callback,
        args=(payload.get('pipelineToken'),
              payload.get('pipelineName'),
              payload.get('data'))
    ).start()
    return jsonify({'status': 'accepted'}), 200

def send_callback(pipeline_token, pipeline_name, data):
    callback_url = 'http://localhost:8080/api/v1/restendpoint/callback'
    requests.post(callback_url, json={
        'pipelineToken': pipeline_token,
        'pipelineName': pipeline_name,
        'data': data
    })

To run:

cd examples/preprocessor
python3 -m venv my-env
source my-env/bin/activate
pip install flask requests
python app.py  # Starts on port 5500

Then configure the pipeline:

"preprocessor": {
  "endpoint": "http://localhost:5500/preprocess/sync",
  "async": false,
  "timeoutMs": 300000
}

Getting Started

Connect your agent

Assistant

Catalog

Taps

Ingestion

Destinations

Data Quality

Transformation

AI Features

Configuration

Governance & Security

Production

Integrations

Recipes

Examples

Processing Flow

Configuration

Fields

Synchronous Mode

Request Payload

Expected Response

Batching

Asynchronous Mode

Flow

Callback Endpoint

Example: Preprocessor Service

​Processing Flow

​Configuration

​Fields

​Synchronous Mode

​Request Payload

​Expected Response

​Batching

​Asynchronous Mode

​Flow

​Callback Endpoint

​Example: Preprocessor Service

Processing Flow

Configuration

Fields

Synchronous Mode

Request Payload

Expected Response

Batching

Asynchronous Mode

Flow

Callback Endpoint

Example: Preprocessor Service