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

Stream execution

Note: /stream-configs/ routes use configuration IDs (config_...). /streams/ routes use stream execution IDs (stream_...).

Understanding IDs

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:

Source options:

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:

Database target spec (db)

Write modes:

Schema policies:

Structure options:

File/S3 target spec

S3 targets add an upload object:

{
  "upload": {
    "bucket": "my-bucket",
    "prefix": "exports/",
    "storageClass": "STANDARD"
  }
}

Snowflake target spec (snowflake)

Everything in the database target spec above, plus staging and, for CDC, currentState.

A CDC stream appends every change to <TABLE>__CHANGES over Snowpipe Streaming, and a Snowflake task merges them into <TABLE> — an ordinary table that holds its own rows. So the stream has no write mode and needs no primary key on the target, and <TABLE>__CHANGES can be trimmed or emptied without touching <TABLE>.

currentState is that task's schedule, written as <number> <unit> with the unit in seconds, minutes, hours or days: 1 minute, 5 minutes, 1 hour. A Snowflake task runs no more often than once a minute and no less often than every 8 days, and any other value — including view, which earlier versions accepted — is refused before the stream starts.

It does not affect how fast changes arrive: they reach <TABLE>__CHANGES within seconds whatever it says. It decides how often a warehouse is woken, and Snowflake bills a minimum of one minute each time it is.

CDC also needs the connecting role to hold EXECUTE TASK on the account, which only an account administrator can grant.

{
  "target": {
    "id": "conn_snowflake",
    "spec": {
      "snowflake": {
        "database": "ANALYTICS",
        "schema": "PUBLIC",
        "schemaPolicy": "create_missing_only",
        "currentState": "5 minutes",
        "staging": { "fileFormat": "parquet" }
      }
    }
  }
}

CDC into Snowflake also requires a key-pair connection and a primary key on every replicated table.

Other configuration fields

Limits:

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 object
• PUT /stream-configs/{id} — updates a configuration, returns the updated object
• DELETE /stream-configs/{id} — requires X-Confirm-Delete: true header, 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
  }
}

• bootstrap appears when source.options.initialSnapshot.enabled = true and bootstrap state has been persisted.
• snapshotCopy appears when load mode (API mode=load) or MySQL CDC bootstrap is actively using resumable chunked snapshot copy for at least one eligible table.

Stream states

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:

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