Define a streaming pipeline over HTTP

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. 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

POST /api/v1/sessions
Host: <ENGINE_HOST>

From the response, keep session_id from /session_id. Endpoint reference.

2. Register the Kafka source

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.

3. Follow: Register the Kafka source

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.

4. Register the Iceberg sink

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.

5. Follow: Register the Iceberg sink

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.

6. Create and start the pipeline

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.

7. Follow: Create and start the pipeline

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.

8. Read: Create and start the pipeline

GET /api/v1/sessions/{{session_id}}/operations/{{statement_2}}/results?limit=1
Host: <ENGINE_HOST>

Endpoint reference.

9. Inspect the continuous pipeline

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.

10. Follow: Inspect the continuous pipeline

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.

11. Read: Inspect the continuous pipeline

GET /api/v1/sessions/{{session_id}}/operations/{{statement_3}}/results?limit=1
Host: <ENGINE_HOST>

Endpoint reference.

12. Stop continuous execution

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.

13. Follow: Stop continuous execution

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.

14. Read: Stop continuous execution

GET /api/v1/sessions/{{session_id}}/operations/{{statement_4}}/results?limit=1
Host: <ENGINE_HOST>

Endpoint reference.

15. Close the session

DELETE /api/v1/sessions/{{session_id}}
Host: <ENGINE_HOST>

Endpoint reference.

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 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.

Transformation Engine

Learn Laminar

Workloads

SQL reference

HTTP API

Recipes

Operations

Define a streaming pipeline over HTTP

Before you start

Define Kafka to Iceberg through HTTP

1. Open a session

2. Register the Kafka source

3. Follow: Register the Kafka source

4. Register the Iceberg sink

5. Follow: Register the Iceberg sink

6. Create and start the pipeline

7. Follow: Create and start the pipeline

8. Read: Create and start the pipeline

9. Inspect the continuous pipeline

10. Follow: Inspect the continuous pipeline

11. Read: Inspect the continuous pipeline

12. Stop continuous execution

13. Follow: Stop continuous execution

14. Read: Stop continuous execution

15. Close the session

Configure and send each statement

Inspect and verify

Stop deliberately

​Before you start

​Define Kafka to Iceberg through HTTP

​1. Open a session

​2. Register the Kafka source

​3. Follow: Register the Kafka source

​4. Register the Iceberg sink

​5. Follow: Register the Iceberg sink

​6. Create and start the pipeline

​7. Follow: Create and start the pipeline

​8. Read: Create and start the pipeline

​9. Inspect the continuous pipeline

​10. Follow: Inspect the continuous pipeline

​11. Read: Inspect the continuous pipeline

​12. Stop continuous execution

​13. Follow: Stop continuous execution

​14. Read: Stop continuous execution

​15. Close the session

​Configure and send each statement

​Inspect and verify

​Stop deliberately

Before you start

Define Kafka to Iceberg through HTTP

1. Open a session

2. Register the Kafka source

3. Follow: Register the Kafka source

4. Register the Iceberg sink

5. Follow: Register the Iceberg sink

6. Create and start the pipeline

7. Follow: Create and start the pipeline

8. Read: Create and start the pipeline

9. Inspect the continuous pipeline

10. Follow: Inspect the continuous pipeline

11. Read: Inspect the continuous pipeline

12. Stop continuous execution

13. Follow: Stop continuous execution

14. Read: Stop continuous execution

15. Close the session

Configure and send each statement

Inspect and verify

Stop deliberately