Streams API Workflows
Use this page when you need to configure and run data streams through the API.
A stream configuration defines what data to move and how — which source connections, tables, target, and write behavior. Starting a configuration creates a stream execution that can be paused, resumed, or stopped.
Authentication
All endpoints require:
-H "X-API-Key: $API_KEY" \
-H "X-Install-ID: $INSTALL_ID"
Endpoints overview
Stream configurations
| Route | Method | Description |
|---|---|---|
/stream-configs | POST | Create a configuration |
/stream-configs | GET | List all configurations |
/stream-configs/{id} | GET | Get configuration details |
/stream-configs/{id} | PUT | Update a configuration |
/stream-configs/{id} | DELETE | Delete a configuration |
/stream-configs/{id}/clone | PUT | Clone a configuration |
/stream-configs/{id}/start | POST | Start a stream |
/stream-configs/{id}/history | GET | Get run history |
Stream execution
| Route | Method | Description |
|---|---|---|
/streams/{id}/pause | POST | Pause a running stream |
/streams/{id}/resume | POST | Resume a paused stream |
/streams/{id}/stop | POST | Stop a stream |
/streams/{id}/stats | GET | Get stream statistics |
/streams/{id}/logs | GET | Get persisted stream logs |
Note: /stream-configs/ routes use configuration IDs (config_...). /streams/ routes use stream execution IDs (stream_...).
Understanding IDs
| ID prefix | Example | Used with |
|---|---|---|
conn_ | conn_2sZyDMeIbdaXnN9OL7vYzzAiZU5 | Connection endpoints |
config_ | config_3B54E8EctFDkIfPiNTqhIogCp5Z | /stream-configs/ |
stream_ | stream_3B54FGHiJklMnOpQrStUvWxYz12 | /streams/ |
Starting a configuration creates a new stream_ ID for that execution. The configuration remains unchanged and can be used to start multiple streams.
Create a stream configuration
POST /stream-configs
Database to database (load mode; API value: load)
curl -X POST "$API_URL/stream-configs" \
-H "X-API-Key: $API_KEY" \
-H "X-Install-ID: $INSTALL_ID" \
-H "Content-Type: application/json" \
-d '{
"name": "mysql_to_postgresql",
"mode": "load",
"source": {
"connections": [
{
"connectionId": "conn_source_id",
"database": "sakila",
"tables": [
{"name": "actor"},
{"name": "film"}
]
}
]
},
"target": {
"id": "conn_target_id",
"spec": {
"db": {
"database": "postgres",
"writeMode": "truncate_and_load",
"structureOptions": {
"tables": true,
"indexes": true,
"foreignKeys": true
}
}
}
}
}'
Database to database (CDC mode)
{
"name": "mysql_cdc_to_pg",
"mode": "cdc",
"source": {
"connections": [
{
"connectionId": "conn_source_id",
"database": "production",
"tables": [
{"name": "orders"},
{"name": "customers"}
]
}
],
"options": {
"operations": ["INSERT", "UPDATE", "DELETE"]
}
},
"target": {
"id": "conn_target_id",
"spec": {
"db": {
"database": "replica",
"writeMode": "upsert",
"schemaPolicy": "validate_existing"
}
}
}
}
Database to S3
{
"name": "mysql_to_s3_parquet",
"mode": "load",
"source": {
"connections": [
{
"connectionId": "conn_mysql_id",
"database": "analytics",
"tables": [{"name": "events"}]
}
]
},
"target": {
"id": "conn_s3_id",
"spec": {
"s3": {
"fileFormat": "parquet",
"upload": {
"bucket": "data-lake",
"prefix": "exports/"
}
}
}
}
}
Database to local files
{
"name": "pg_to_csv",
"mode": "load",
"source": {
"connections": [
{
"connectionId": "conn_pg_id",
"database": "reporting",
"tables": [{"name": "monthly_report"}]
}
]
},
"target": {
"id": "conn_files_id",
"spec": {
"files": {
"fileFormat": "csv",
"format": {
"compression": "gzip"
}
}
}
}
}
Response
Returns the created configuration with a generated config_ ID:
{
"id": "config_3B54E8EctFDkIfPiNTqhIogCp5Z",
"name": "mysql_to_postgresql",
"mode": "load",
"source": {
"connections": [
{
"connectionId": "conn_source_id",
"database": "sakila",
"tables": [
{"name": "actor"},
{"name": "film"}
]
}
]
},
"target": {
"id": "conn_target_id",
"spec": {
"db": {
"database": "postgres",
"writeMode": "truncate_and_load",
"structureOptions": {
"tables": true,
"indexes": true,
"foreignKeys": true
}
}
}
},
"created": 1773765482,
"reportingInterval": 3
}
Configuration fields
Source
{
"source": {
"connections": [{...}],
"options": {...}
}
}
Each entry in connections specifies:
| Field | Type | Description |
|---|---|---|
connectionId | string | Connection ID (required) |
database | string | Database name |
schema | string | Schema name (optional, PostgreSQL/Snowflake) |
tables | array | Tables to replicate |
queries | array | Virtual query sources (load mode only; API mode=load) |
s3 | object | S3 source config (mutually exclusive with tables) |
files | object | File source config (mutually exclusive with tables) |
Source options:
| Field | Type | Description |
|---|---|---|
dataBundleSize | number | Rows per NATS message (default: 500) |
operations | array | CDC operations filter: INSERT, UPDATE, DELETE |
replicationSlot | string | PostgreSQL CDC replication slot name (default: {database}_dbconvert_replication_slot) |
publicationName | string | PostgreSQL CDC publication name (default: dbconvert-publication) |
initialSnapshot | object | CDC-only bootstrap option. Set enabled: true to copy existing rows before CDC starts. |
source.options.initialSnapshot.restartPolicy currently supports fail_until_reset.
Table selection
Tables support optional structured selection for filtering and column selection:
{
"name": "orders",
"selection": {
"selectedColumns": ["id", "customer_id", "total"],
"filters": [
{"column": "status", "operator": "=", "value": "active"}
],
"sorts": [
{"column": "created_at", "direction": "DESC"}
],
"limit": 10000
}
}
Target
{
"target": {
"id": "conn_target_id",
"spec": {
"db": {...}
}
}
}
target.id is the connection ID. target.spec uses one variant matching the target type:
| Spec field | Target type |
|---|---|
db | MySQL, PostgreSQL, Snowflake (data) |
files | Local file system |
s3 | Amazon S3 |
gcs | Google Cloud Storage |
azure | Azure Blob Storage |
snowflake | Snowflake (with staging) |
Database target spec (db)
| Field | Type | Description |
|---|---|---|
database | string | Target database name (required) |
schema | string | Target schema (optional) |
writeMode | string | How to handle existing data |
schemaPolicy | string | How to handle existing schema |
structureOptions | object | Which DDL to create (tables, indexes, etc.) |
skipData | boolean | Create structure only, skip data transfer |
Write modes:
| Value | Description |
|---|---|
fail_if_not_empty | Error if target table has data (default for load mode, API mode=load) |
append | Add rows to existing data |
truncate_and_load | Clear target tables before writing |
upsert | Insert or update on conflict (default for CDC) |
Schema policies:
| Value | Description |
|---|---|
fail_if_exists | Error if target tables exist (default for load mode, API mode=load) |
validate_existing | Use existing tables, validate schema (default for CDC) |
create_missing_only | Create only missing tables |
drop_and_recreate | Drop and recreate all target tables |
Structure options:
| Field | Type | Description |
|---|---|---|
tables | boolean | Create tables |
indexes | boolean | Create indexes |
foreignKeys | boolean | Create foreign key constraints |
checkConstraints | boolean | Create check constraints |
File/S3 target spec
| Field | Type | Description |
|---|---|---|
fileFormat | string | csv, json, jsonl, or parquet |
format | object | Format-specific settings (compression, etc.) |
outputDirectory | string | Local staging path (optional) |
S3 targets add an upload object:
{
"upload": {
"bucket": "my-bucket",
"prefix": "exports/",
"storageClass": "STANDARD"
}
}
Other configuration fields
| Field | Type | Description |
|---|---|---|
name | string | Configuration name (required) |
mode | string | convert or cdc (required) |
reportingInterval | number | Stats reporting interval in seconds |
limits | object | Optional execution limits |
description | string | Optional description |
Limits:
| Field | Type | Description |
|---|---|---|
numberOfEvents | number | Max rows to process (0 = unlimited) |
elapsedTime | number | Max seconds to run (0 = unlimited) |
List stream configurations
GET /stream-configs
Returns a wrapped response. Supports search, limit, and offset query parameters.
curl "$API_URL/stream-configs" \
-H "X-API-Key: $API_KEY" \
-H "X-Install-ID: $INSTALL_ID"
Response:
{
"items": [...],
"total": 93,
"filtered": 93
}
Get, update, delete
GET /stream-configs/{id}— returns the full configuration objectPUT /stream-configs/{id}— updates a configuration, returns the updated objectDELETE /stream-configs/{id}— requiresX-Confirm-Delete: trueheader, returns 204
Clone a configuration
PUT /stream-configs/{id}/clone
Creates a copy with a new ID and _copy appended to the name. Returns the full cloned configuration.
Start a stream
POST /stream-configs/{id}/start
Starts a stream execution from a configuration. Only one stream can be active at a time.
Optional query parameter: ?structureOnly=true to create target structure without transferring data.
Response:
{
"id": "stream_3B54FGHiJklMnOpQrStUvWxYz12"
}
Use the returned stream_ ID for pause, resume, stop, and stats endpoints.
Pause, resume, stop
All return HTTP 204 No Content with no response body.
# Pause a running stream
curl -X POST "$API_URL/streams/$STREAM_ID/pause" \
-H "X-API-Key: $API_KEY" \
-H "X-Install-ID: $INSTALL_ID"
# Resume a paused stream
curl -X POST "$API_URL/streams/$STREAM_ID/resume" \
-H "X-API-Key: $API_KEY" \
-H "X-Install-ID: $INSTALL_ID"
# Stop a stream
curl -X POST "$API_URL/streams/$STREAM_ID/stop" \
-H "X-API-Key: $API_KEY" \
-H "X-Install-ID: $INSTALL_ID"
Stop accepts an optional ?force=true query parameter for recovery scenarios.
Get stream statistics
GET /streams/{id}/stats
Returns current execution state and per-node counters. For CDC and chunked snapshot workflows, the same response can also include reliability, bootstrap, and snapshotCopy blocks. See Logs And Monitoring for the full response structure and status values.
curl "$API_URL/streams/$STREAM_ID/stats" \
-H "X-API-Key: $API_KEY" \
-H "X-Install-ID: $INSTALL_ID"
Response:
{
"configID": "config_3B54E8EctFDkIfPiNTqhIogCp5Z",
"streamID": "stream_3B54FGHiJklMnOpQrStUvWxYz12",
"status": "RUNNING",
"nodes": [
{
"id": "reader-1",
"type": "source",
"stat": {
"counter": 15000,
"failedCounter": 0,
"sumDataSize": 2048576,
"avgRate": 17067,
"status": "RUNNING"
}
}
]
}
When present, the optional workflow blocks look like this:
{
"bootstrap": {
"engine": "mysql",
"phase": "copying",
"handoffPosition": "mysql-bin.000176:79269"
},
"snapshotCopy": {
"status": "copying",
"mode": "cdc_bootstrap",
"strategy": "pk_range",
"tablesPlanned": 12,
"tablesCompleted": 4,
"chunksPlanned": 96,
"chunksCompleted": 31
}
}
bootstrapappears whensource.options.initialSnapshot.enabled = trueand bootstrap state has been persisted.snapshotCopyappears when load mode (APImode=load) or MySQL CDC bootstrap is actively using resumable chunked snapshot copy for at least one eligible table.
Stream states
| Status | Description |
|---|---|
RUNNING | Actively processing data |
PAUSED | Temporarily paused, can be resumed |
FINISHED | Completed successfully |
FAILED | Terminated with an error |
STOPPED | Manually stopped |
TIME_LIMIT_REACHED | Stopped after reaching time limit |
EVENT_LIMIT_REACHED | Stopped after reaching event limit |
Error handling
{
"error": "Cannot start new stream: found active streams that need attention",
"type": "validation",
"code": "VALIDATION_FAILED",
"details": {
"field": "stream_state",
"reason": "Cannot start new stream: found active streams that need attention"
}
}
Common status codes:
| Code | Meaning |
|---|---|
| 400 | Validation error (missing fields, invalid config) |
| 401 | Invalid or missing API key |
| 404 | Configuration or stream not found |
| 503 | Service error |
Related docs
- Stream API Examples — Programming examples
- Connections API — Connection management
- Logs And Monitoring — Stream stats and log retrieval
- SQL Logs — Live SQL query visibility