> ## Documentation Index
> Fetch the complete documentation index at: https://docs.e6data.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Define a streaming pipeline over HTTP

> Register a Kafka source, register an Iceberg sink, and create a continuous pipeline through SQL session operations. Execution uses the native streaming path.

How-to guideApplies to: Unified engine deployments

Register a Kafka source, register an Iceberg sink, and create a continuous pipeline through SQL session operations. Execution uses the native streaming path.

## Before you start

Complete [Streaming setup](https://docs.e6data.com/transformation/streaming/before-you-start). You need Kafka, an accessible Iceberg REST catalog and warehouse, configured storage access, and streaming workers. Choose unused source, sink, pipeline, and destination names. Replace the practice names throughout if they already exist.

### Define Kafka to Iceberg through HTTP

Send each statement separately in one session. Wait for success before its dependent step. Cleanup is explicit.

| Placeholder | Value |
| - | - |
| `{{kafka_brokers}}` | Your environment value. |
| `{{topic}}` | `"orders"` |
| `{{iceberg_rest_url}}` | Your environment value. |
| `{{warehouse}}` | Your environment value. |
| `{{statement_0}}` | A new UUID for this request; reuse it when recovering the same submission. |
| `{{statement_1}}` | A new UUID for this request; reuse it when recovering the same submission. |
| `{{statement_2}}` | A new UUID for this request; reuse it when recovering the same submission. |
| `{{statement_3}}` | A new UUID for this request; reuse it when recovering the same submission. |
| `{{statement_4}}` | A new UUID for this request; reuse it when recovering the same submission. |

Replace the placeholders in your client. Keep returned IDs for later requests. Paths below use the native API prefix; adjust it for your deployment.

#### 1. Open a session

```http theme={"theme":{"light":"github-light","dark":"github-dark"}}
POST /api/v1/sessions
Host: <ENGINE_HOST>
```

From the response, keep `session_id` from `/session_id`.

[Endpoint reference](https://docs.e6data.com/transformation/http/reference/open-session.md).

#### 2. Register the Kafka source

```http theme={"theme":{"light":"github-light","dark":"github-dark"}}
POST /api/v1/sessions/{{session_id}}/operations
Host: <ENGINE_HOST>
Content-Type: application/json

{
  "operation_id": "{{statement_0}}",
  "sql": "CREATE TABLE http_docs_orders_source (order_id BIGINT, customer_id BIGINT, amount DOUBLE, status TEXT) WITH (connector = 'kafka', type = 'source', bootstrap_servers = '{{kafka_brokers}}', topic = '{{topic}}', format = 'json', 'source.offset' = 'earliest', 'source.read_mode' = 'read_committed')",
  "source": "external"
}
```

[Endpoint reference](https://docs.e6data.com/transformation/http/reference/submit-operation.md).

#### 3. Follow: Register the Kafka source

```http theme={"theme":{"light":"github-light","dark":"github-dark"}}
GET /api/v1/sessions/{{session_id}}/operations/{{statement_0}}
Host: <ENGINE_HOST>
```

Repeat this read until `/status` reaches a terminal state: `Succeeded`, `Failed`, `Cancelled`, `OutcomeUnknown`, `Skipped`.

[Endpoint reference](https://docs.e6data.com/transformation/http/reference/operation-status.md).

#### 4. Register the Iceberg sink

```http theme={"theme":{"light":"github-light","dark":"github-dark"}}
POST /api/v1/sessions/{{session_id}}/operations
Host: <ENGINE_HOST>
Content-Type: application/json

{
  "operation_id": "{{statement_1}}",
  "sql": "CREATE TABLE http_docs_paid_sink WITH (connector = 'iceberg', type = 'sink', 'catalog.type' = 'rest', 'catalog.url' = '{{iceberg_rest_url}}', 'catalog.warehouse' = '{{warehouse}}', namespace = 'analytics', table = 'http_docs_paid_orders', append_only = 'true', infer_schema = 'true')",
  "source": "external"
}
```

[Endpoint reference](https://docs.e6data.com/transformation/http/reference/submit-operation.md).

#### 5. Follow: Register the Iceberg sink

```http theme={"theme":{"light":"github-light","dark":"github-dark"}}
GET /api/v1/sessions/{{session_id}}/operations/{{statement_1}}
Host: <ENGINE_HOST>
```

Repeat this read until `/status` reaches a terminal state: `Succeeded`, `Failed`, `Cancelled`, `OutcomeUnknown`, `Skipped`.

[Endpoint reference](https://docs.e6data.com/transformation/http/reference/operation-status.md).

#### 6. Create and start the pipeline

```http theme={"theme":{"light":"github-light","dark":"github-dark"}}
POST /api/v1/sessions/{{session_id}}/operations
Host: <ENGINE_HOST>
Content-Type: application/json

{
  "operation_id": "{{statement_2}}",
  "sql": "CREATE STREAMING PIPELINE http_docs_orders_to_iceberg WITH (parallelism = 1, checkpoint_interval = '10 seconds') AS INSERT INTO http_docs_paid_sink SELECT order_id, customer_id, amount, amount * 1.10 AS amount_with_tax FROM http_docs_orders_source WHERE status = 'paid' AND amount > 0",
  "source": "external"
}
```

[Endpoint reference](https://docs.e6data.com/transformation/http/reference/submit-operation.md).

#### 7. Follow: Create and start the pipeline

```http theme={"theme":{"light":"github-light","dark":"github-dark"}}
GET /api/v1/sessions/{{session_id}}/operations/{{statement_2}}
Host: <ENGINE_HOST>
```

Repeat this read until `/status` reaches a terminal state: `Succeeded`, `Failed`, `Cancelled`, `OutcomeUnknown`, `Skipped`.

[Endpoint reference](https://docs.e6data.com/transformation/http/reference/operation-status.md).

#### 8. Read: Create and start the pipeline

```http theme={"theme":{"light":"github-light","dark":"github-dark"}}
GET /api/v1/sessions/{{session_id}}/operations/{{statement_2}}/results?limit=1
Host: <ENGINE_HOST>
```

[Endpoint reference](https://docs.e6data.com/transformation/http/reference/operation-results.md).

#### 9. Inspect the continuous pipeline

```http theme={"theme":{"light":"github-light","dark":"github-dark"}}
POST /api/v1/sessions/{{session_id}}/operations
Host: <ENGINE_HOST>
Content-Type: application/json

{
  "operation_id": "{{statement_3}}",
  "sql": "DESCRIBE PIPELINE http_docs_orders_to_iceberg",
  "source": "external"
}
```

[Endpoint reference](https://docs.e6data.com/transformation/http/reference/submit-operation.md).

#### 10. Follow: Inspect the continuous pipeline

```http theme={"theme":{"light":"github-light","dark":"github-dark"}}
GET /api/v1/sessions/{{session_id}}/operations/{{statement_3}}
Host: <ENGINE_HOST>
```

Repeat this read until `/status` reaches a terminal state: `Succeeded`, `Failed`, `Cancelled`, `OutcomeUnknown`, `Skipped`.

[Endpoint reference](https://docs.e6data.com/transformation/http/reference/operation-status.md).

#### 11. Read: Inspect the continuous pipeline

```http theme={"theme":{"light":"github-light","dark":"github-dark"}}
GET /api/v1/sessions/{{session_id}}/operations/{{statement_3}}/results?limit=1
Host: <ENGINE_HOST>
```

[Endpoint reference](https://docs.e6data.com/transformation/http/reference/operation-results.md).

#### 12. Stop continuous execution

```http theme={"theme":{"light":"github-light","dark":"github-dark"}}
POST /api/v1/sessions/{{session_id}}/operations
Host: <ENGINE_HOST>
Content-Type: application/json

{
  "operation_id": "{{statement_4}}",
  "sql": "STOP PIPELINE http_docs_orders_to_iceberg",
  "source": "external"
}
```

[Endpoint reference](https://docs.e6data.com/transformation/http/reference/submit-operation.md).

#### 13. Follow: Stop continuous execution

```http theme={"theme":{"light":"github-light","dark":"github-dark"}}
GET /api/v1/sessions/{{session_id}}/operations/{{statement_4}}
Host: <ENGINE_HOST>
```

Repeat this read until `/status` reaches a terminal state: `Succeeded`, `Failed`, `Cancelled`, `OutcomeUnknown`, `Skipped`.

[Endpoint reference](https://docs.e6data.com/transformation/http/reference/operation-status.md).

#### 14. Read: Stop continuous execution

```http theme={"theme":{"light":"github-light","dark":"github-dark"}}
GET /api/v1/sessions/{{session_id}}/operations/{{statement_4}}/results?limit=1
Host: <ENGINE_HOST>
```

[Endpoint reference](https://docs.e6data.com/transformation/http/reference/operation-results.md).

#### 15. Close the session

```http theme={"theme":{"light":"github-light","dark":"github-dark"}}
DELETE /api/v1/sessions/{{session_id}}
Host: <ENGINE_HOST>
```

[Endpoint reference](https://docs.e6data.com/transformation/http/reference/close-session.md).

## Configure and send each statement

Enter the Kafka brokers, topic, catalog URL, and warehouse under **Example values**. The example sends each statement as its own SQL operation in the same session, with a separate operation identity. Wait for each required predecessor to succeed.

Creating the pipeline starts continuous execution by default. A successful create operation means the command was accepted; it does not prove that streaming workers reached Running or that the first checkpoint committed.

## Inspect and verify

Run the DESCRIBE step and inspect the `result` column. To repeat that inspection, prepare a new SQL operation with a fresh identity. Reusing the old identity reads the earlier operation rather than issuing a new DESCRIBE.

Use the [Kafka to Iceberg lesson](https://docs.e6data.com/transformation/streaming/first-pipeline) for sample input records and expected output. Verify the destination after a successful checkpoint. Streaming connector names are separate from the qualified catalog table name used to query Iceberg through Unified.

## Stop deliberately

Send the Stop pipeline statement when you are ready to end continuous execution. Follow its reported state. Stopping a docs watch or closing the HTTP session does not stop the pipeline.

Keep or remove connector definitions and destination objects through their supported management flows. Stopping the pipeline does not delete its published data.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.