Pipelines

Pipelines represent the data flows within a connection, defining how data is collected, transformed, and delivered — for example, by ingesting users, exporting profiles, or routing events.

This section documents the endpoints common to various types of pipelines. Instead, to create, update, or retrieve a pipeline, refer to the specific sections for each pipeline type.

Endpoints

Set status

Sets the status of a pipeline.

Request

:id
int Required
The ID of the pipeline.
enabled
boolean Required
Indicates if the pipeline is enabled.

Response

No response.

PUT /v1/pipelines/:id/status

curl -X PUT https://example.com/v1/pipelines/705981339/status \
  -H "Authorization: Bearer api_xxxxxxx" \
  --json '{
    "enabled": true
  }'

Errors

404

workspace does not exist

404

pipeline does not exist

Set schedule period

Sets the frequency at which a pipeline imports users or export profiles. The pipeline must be enabled for the run to be scheduled. If the period is null, the scheduler will be disabled.

Request

:id
int Required
The ID of the pipeline.
period
nullable string Required
The schedule period. It determines how often the run is triggered automatically. If it is null, the scheduler is disabled and no automatic runs will occur.
Possible values: "5m", "15m", "30m", "1h", "2h", "3h", "6h", "8h", "12h" or "24h".

Response

No response.

PUT /v1/pipelines/:id/schedule

curl -X PUT https://example.com/v1/pipelines/705981339/schedule \
  -H "Authorization: Bearer api_xxxxxxx" \
  --json '{
    "period": "1h"
  }'

Errors

404

workspace does not exist

404

pipeline does not exist

Run pipeline

Starts a pipeline run to import users into the data warehouse or export warehouse profiles for activation. A run applies the pipeline's filters and transformations.

It returns immediately without waiting for the run to complete. To track the progress, call the Get pipeline run endpoint using the returned run ID.

The pipeline must be enabled.

Request

:id
int Required
The ID of the pipeline that imports users or exports profiles.
incremental
nullable boolean
Determines whether users should be imported from scratch (false) or incrementally from the previous import (true). If omitted or nil, the pipeline's default setting will be used.

For source database and file pipelines, the pipeline must include a "update time" property.

For destination pipelines, this parameter must be omitted or set to nil.

Response

id
int
The ID of the run that was started.

POST /v1/pipelines/:id/runs

curl https://example.com/v1/pipelines/705981339/runs \
  -H "Authorization: Bearer api_xxxxxxx" \
  --json '{
    "incremental": true
  }'

Response

{
  "id": 609461413
}

Errors

404

workspace does not exist

404

pipeline does not exist

422

pipeline is disabled

422

incremental requires an update time column

422

pipeline is already in progress

422

data warehouse is in inspection mode

422

data warehouse is in maintenance mode

List pipeline runs

Returns all pipeline runs.

Pipeline runs are automatically triggered by the scheduler or can be started by calling the specific endpoint for the pipeline.

Request

No parameters.

Response

runs
array of object
The pipeline runs.
- id
  int
  The ID of the run.
- pipeline
  int
  The ID of the pipeline that was run.
- startTime
  datetime
  The start time in ISO 8601 format.
- endTime
  nullable datetime
  The end time in ISO 8601 format. It is null if the run has not yet finished.
- passed
  array of int
  The number of users or events that successfully passed each step. All values will be 0 if the run has not yet finished.
- failed
  array of int
  The number of users or events that failed at each step. All values will be 0 if the run has not yet finished.
- error
  string
  An error occurred during the run, causing it to stop prematurely. It is empty if the run has not yet finished or if no error occurred.

GET /v1/pipelines/runs

curl https://example.com/v1/pipelines/runs \
  -H "Authorization: Bearer api_xxxxxxx"

Response

{
  "runs": [
    {
      "id": 609461413,
      "pipeline": 705981339,
      "startTime": "2024-11-27T18:22:47.937Z",
      "endTime": "2024-11-27T18:49:07.150Z",
      "passed": [
        6029,
        6029,
        5974,
        5974,
        5974,
        5974
      ],
      "failed": [
        0,
        0,
        55,
        0,
        0,
        0
      ],
      "error": ""
    }
  ]
}

Errors

404

workspace does not exist

Get pipeline run

Returns a pipeline run.

Pipeline runs are automatically triggered by the scheduler or can be started by calling the specific endpoint for the pipeline.

Request

:id
int Required
The ID of the run.

Response

id
int
The ID of the run.
pipeline
int
The ID of the pipeline that was run.
startTime
datetime
The start time in ISO 8601 format.
endTime
nullable datetime
The end time in ISO 8601 format. It is null if the run has not yet finished.
passed
array of int
The number of users or events that successfully passed each step. All values will be 0 if the run has not yet finished.
failed
array of int
The number of users or events that failed at each step. All values will be 0 if the run has not yet finished.
error
string
An error occurred during the run, causing it to stop prematurely. It is empty if the run has not yet finished or if no error occurred.

GET /v1/pipelines/runs/:id

curl https://example.com/v1/pipelines/runs/609461413 \
  -H "Authorization: Bearer api_xxxxxxx"

Response

{
  "id": 609461413,
  "pipeline": 705981339,
  "startTime": "2024-11-27T18:22:47.937Z",
  "endTime": "2024-11-27T18:49:07.150Z",
  "passed": [
    6029,
    6029,
    5974,
    5974,
    5974,
    5974
  ],
  "failed": [
    0,
    0,
    55,
    0,
    0,
    0
  ],
  "error": ""
}

Errors

404

workspace does not exist

Delete pipeline

Delete a pipeline.

Request

:id
int Required
The ID of the pipeline to delete.

Response

No response.

DELETE /v1/pipelines/:id

curl -X DELETE https://example.com/v1/pipelines/705981339 \
  -H "Authorization: Bearer api_xxxxxxx"

Errors

404

workspace does not exist

404

pipeline does not exist