HTTP API - e6data Docs

Base URL

http://<controller-host>:8000/api/v1

All endpoints below are relative to this base URL.

Authentication

Include the API token in the Authorization header:

curl -H "Authorization: Bearer <token>" http://localhost:8000/api/v1/pipelines

Response Format

All responses are JSON. Errors return:

{
  "error": "descriptive error message"
}

List endpoints use cursor-based pagination:

{
  "data": [...],
  "has_more": true
}

Pass ?starting_after=<last_id>&limit=10 to paginate.

Pipelines

Create Pipeline

POST /pipelines

Creates a pipeline and starts it immediately. Request:

{
  "name": "my-pipeline",
  "query": "INSERT INTO sink SELECT * FROM source",
  "parallelism": 4,
  "checkpoint_interval_micros": 30000000
}

Field	Type	Required	Description
`name`	string	yes	Pipeline name
`query`	string	yes	SQL query defining the pipeline
`parallelism`	integer	yes	Number of parallel subtasks
`checkpoint_interval_micros`	integer	no	Checkpoint interval in microseconds (default: 30s)

Response: Pipeline object

Create Preview Pipeline

POST /pipelines/preview

Creates a short-lived pipeline (60s TTL) for testing queries interactively. Request:

{
  "query": "SELECT * FROM source WHERE amount > 100",
  "enable_sinks": false
}

Field	Type	Required	Description
`query`	string	yes	SQL query to preview
`enable_sinks`	boolean	no	Whether to write to real sinks (default: false)

Response: Pipeline object

List Pipelines

GET /pipelines

Query Parameters:

Parameter	Type	Description
`starting_after`	string	Cursor for pagination
`limit`	integer	Items per page (default: 10)

Response:

{
  "data": [Pipeline, ...],
  "has_more": false
}

Get Pipeline

GET /pipelines/:id

Response: Pipeline object

{
  "id": "abc123",
  "name": "my-pipeline",
  "query": "INSERT INTO sink SELECT * FROM source",
  "parallelism": 4,
  "checkpoint_interval_micros": 30000000,
  "stop": "none",
  "created_at": 1700000000000000,
  "graph": {
    "nodes": [...],
    "edges": [...]
  },
  "action": "stop",
  "action_text": "Stop",
  "action_in_progress": false,
  "preview": false
}

Update Pipeline

PATCH /pipelines/:id

Updates pipeline configuration. Only provided fields are changed. Request:

{
  "parallelism": 8,
  "checkpoint_interval_micros": 60000000,
  "stop": "checkpoint"
}

Field	Type	Description
`parallelism`	integer	New parallelism (triggers rescaling)
`checkpoint_interval_micros`	integer	New checkpoint interval
`stop`	string	Stop mode: `"none"`, `"graceful"`, `"immediate"`, `"checkpoint"`, `"force"`

Response: Pipeline object

Restart Pipeline

POST /pipelines/:id/restart

Restarts a stopped or failed pipeline. Request:

{
  "force": false
}

Field	Type	Description
`force`	boolean	Skip taking a final checkpoint before restart (default: false)

Response: Pipeline object

Delete Pipeline

DELETE /pipelines/:id

Deletes a pipeline. The pipeline must be in a terminal state (Stopped, Finished, or Failed). Response: 200 OK

Validate Query

POST /pipelines/validate_query

Validates a SQL query without creating a pipeline. Returns the execution graph or validation errors. Request:

{
  "query": "INSERT INTO sink SELECT * FROM source"
}

Response:

{
  "graph": {
    "nodes": [...],
    "edges": [...]
  },
  "errors": []
}

Jobs

A job represents a single execution of a pipeline. Each time a pipeline starts or restarts, a new job is created.

List Jobs

GET /jobs

Returns all jobs across all pipelines.

GET /pipelines/:pipeline_id/jobs

Returns jobs for a specific pipeline. Response:

{
  "data": [
    {
      "id": 1,
      "running_desired": true,
      "state": "Running",
      "run_id": 3,
      "start_time": 1700000000000000,
      "finish_time": null,
      "tasks": 8,
      "failure_message": null,
      "created_at": 1700000000000000
    }
  ]
}

Get Job Errors

GET /pipelines/:pipeline_id/jobs/:job_id/errors

Returns error and warning messages for a job. Query Parameters:

Parameter	Type	Description
`starting_after`	string	Cursor for pagination
`limit`	integer	Items per page

Response:

{
  "data": [
    {
      "id": "msg-123",
      "created_at": 1700000000000000,
      "operator_id": "operator_2",
      "task_index": 0,
      "level": "Error",
      "message": "Connection refused",
      "details": "..."
    }
  ],
  "has_more": false
}

Get Job Checkpoints

GET /pipelines/:pipeline_id/jobs/:job_id/checkpoints

Returns all checkpoints for a job. Response:

{
  "data": [
    {
      "epoch": 5,
      "backend": "s3",
      "start_time": 1700000150000000,
      "finish_time": 1700000151000000
    }
  ]
}

Get Checkpoint Details

GET /pipelines/:pipeline_id/jobs/:job_id/checkpoints/:epoch/operator_checkpoint_groups

Returns per-operator checkpoint details for a specific epoch. Response:

{
  "data": [
    {
      "operator_id": "operator_0",
      "bytes": 4096,
      "finish_time": 1700000151000000,
      "subtasks": [...]
    }
  ]
}

Get Job Metrics

GET /pipelines/:pipeline_id/jobs/:job_id/operator_metric_groups

Returns per-operator metrics for a running job.

Get Job Output (SSE)

GET /pipelines/:pipeline_id/jobs/:job_id/output

Streams pipeline output in real-time via Server-Sent Events. Only available for pipelines with a preview sink. Response: text/event-stream Each event contains:

{
  "column_names": ["id", "name", "amount"],
  "data": [["1", "Alice", "100"], ["2", "Bob", "200"]]
}

Connection Profiles

A connection profile stores reusable connection credentials (e.g., Kafka bootstrap servers, authentication).

Create Connection Profile

POST /connection_profiles

Request:

{
  "name": "production-kafka",
  "connector": "kafka",
  "config": {
    "bootstrap_servers": "kafka-1:9092,kafka-2:9092",
    "authentication": {
      "type": "sasl_plain",
      "username": "user",
      "password": "pass"
    }
  }
}

Response: ConnectionProfile object Typed endpoints are also available for specific connectors:

POST /connectors/kafka/profiles
POST /connectors/confluent/profiles

List Connection Profiles

GET /connection_profiles

Response:

{
  "data": [
    {
      "id": "prof-123",
      "name": "production-kafka",
      "connector": "kafka",
      "config": {...},
      "description": "Kafka cluster"
    }
  ]
}

Get Connection Profile

GET /connection_profiles/:id

Delete Connection Profile

DELETE /connection_profiles/:id

Fails if the profile is referenced by any connection tables.

Test Connection Profile

POST /connection_profiles/test

Tests connectivity with the provided configuration. Request: Same as create. Response:

{
  "message": "Successfully connected",
  "done": true
}

Autocomplete

GET /connection_profiles/:id/autocomplete

Returns available values for the connection (e.g., Kafka topic names). Response:

{
  "values": {
    "topics": ["orders", "users", "events"]
  }
}

Connection Tables

A connection table defines a specific source or sink - binding a connector, connection profile, schema, and table-specific configuration together.

Create Connection Table

POST /connection_tables

Request:

{
  "name": "orders",
  "connector": "kafka",
  "connection_profile_id": "prof-123",
  "config": {
    "topic": "orders",
    "read_mode": "read_uncommitted"
  },
  "schema": {
    "format": { "json": {} },
    "fields": [
      { "field_name": "id", "field_type": { "primitive": "Int64" } },
      { "field_name": "amount", "field_type": { "primitive": "Float64" } }
    ]
  }
}

Typed endpoints are also available:

POST /connectors/kafka/tables
POST /connectors/confluent/tables
POST /connectors/cdc/tables
POST /connectors/iceberg/tables
POST /connectors/delta/tables
POST /connectors/filesystem/tables
POST /connectors/kinesis/tables
POST /connectors/http_source/tables
POST /connectors/stdout/tables

Response: ConnectionTable object

{
  "id": 1,
  "pub_id": "tbl-456",
  "name": "orders",
  "created_at": 1700000000000000,
  "connector": "kafka",
  "table_type": "source",
  "config": {...},
  "schema": {...},
  "consumers": 1
}

List Connection Tables

GET /connection_tables

Query Parameters:

Parameter	Type	Description
`starting_after`	string	Cursor for pagination
`limit`	integer	Items per page

Get Connection Table

GET /connection_tables/:id

Delete Connection Table

DELETE /connection_tables/:id

Test Connection Table

POST /connection_tables/test

Tests reading from a source with the provided configuration. Returns results as a Server-Sent Events stream.

Validate Schema

POST /connection_tables/schemas/test

Validates a schema definition.

Connectors

List Available Connectors

GET /connectors

Returns metadata about all available connectors. Response:

{
  "data": [
    {
      "name": "kafka",
      "description": "Apache Kafka source and sink",
      "connection_config": {...},
      "table_config": {...}
    }
  ]
}

Health

Ping

GET /ping

Returns "Pong" - used for health checks. No authentication required.

​Base URL

​Authentication

​Response Format

​Pipelines

​Create Pipeline

​Create Preview Pipeline

​List Pipelines

​Get Pipeline

​Update Pipeline

​Restart Pipeline

​Delete Pipeline

​Validate Query

​Jobs

​List Jobs

​Get Job Errors

​Get Job Checkpoints

​Get Checkpoint Details

​Get Job Metrics

​Get Job Output (SSE)

​Connection Profiles

​Create Connection Profile

​List Connection Profiles

​Get Connection Profile

​Delete Connection Profile

​Test Connection Profile

​Autocomplete

​Connection Tables

​Create Connection Table

​List Connection Tables

​Get Connection Table

​Delete Connection Table

​Test Connection Table

​Validate Schema

​Connectors

​List Available Connectors

​Health

​Ping

Base URL

Authentication

Response Format

Pipelines

Create Pipeline

Create Preview Pipeline

List Pipelines

Get Pipeline

Update Pipeline

Restart Pipeline

Delete Pipeline

Validate Query

Jobs

List Jobs

Get Job Errors

Get Job Checkpoints

Get Checkpoint Details

Get Job Metrics

Get Job Output (SSE)

Connection Profiles

Create Connection Profile

List Connection Profiles

Get Connection Profile

Delete Connection Profile

Test Connection Profile

Autocomplete

Connection Tables

Create Connection Table

List Connection Tables

Get Connection Table

Delete Connection Table

Test Connection Table

Validate Schema

Connectors

List Available Connectors

Health

Ping