> For the complete documentation index, see [llms.txt](https://docs.e6data.com/ingestion-engine/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.e6data.com/ingestion-engine/api/api.md). # HTTP API e6 Ingestion Engine exposes an HTTP API for managing pipelines, connections, and jobs programmatically. ## Base URL ``` http://:8000/api/v1 ``` All endpoints below are relative to this base URL. ## Authentication Include the API token in the `Authorization` header: ```bash curl -H "Authorization: Bearer " http://localhost:8000/api/v1/pipelines ``` ## Response Format All responses are JSON. Errors return: ```json { "error": "descriptive error message" } ``` List endpoints use cursor-based pagination: ```json { "data": [...], "has_more": true } ``` Pass `?starting_after=&limit=10` to paginate. *** ## Pipelines ### Create Pipeline ``` POST /pipelines ``` Creates a pipeline and starts it immediately. **Request:** ```json { "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:** ```json { "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:** ```json { "data": [Pipeline, ...], "has_more": false } ``` *** ### Get Pipeline ``` GET /pipelines/:id ``` **Response:** `Pipeline` object ```json { "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:** ```json { "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:** ```json { "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:** ```json { "query": "INSERT INTO sink SELECT * FROM source" } ``` **Response:** ```json { "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:** ```json { "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:** ```json { "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:** ```json { "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:** ```json { "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: ```json { "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:** ```json { "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:** ```json { "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:** ```json { "message": "Successfully connected", "done": true } ``` *** ### Autocomplete ``` GET /connection_profiles/:id/autocomplete ``` Returns available values for the connection (e.g., Kafka topic names). **Response:** ```json { "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:** ```json { "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/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 ```json { "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:** ```json { "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. --- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://docs.e6data.com/ingestion-engine/api/api.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.