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

# Your first pipeline

> Read JSON orders from Kafka, keep paid orders with a positive amount, and write them to Iceberg. The complete pipeline definition is SQL.

TutorialApplies to: Unified engine deployments

Read JSON orders from Kafka, keep paid orders with a positive amount, and write them to Iceberg. The complete pipeline definition is SQL.

To submit these statements through HTTP, follow [Define a streaming pipeline over HTTP](https://docs.e6data.com/transformation/http/examples/streaming.md).

> **Before you run:** complete [Before you start](https://docs.e6data.com/transformation/streaming/before-you-start). Replace the broker, catalog, and warehouse values below. Use an empty test topic and an unused destination table if you want to compare the exact example result.

## 1. Define the source, sink, and pipeline

Copy this complete script into Workbench. Update its connection values and run the statements in order. Register the source and sink once. The third statement creates and starts the pipeline.

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
CREATE TABLE orders_source (
  order_id BIGINT,
  customer_id BIGINT,
  amount DOUBLE,
  status TEXT
) WITH (
  connector = 'kafka',
  type = 'source',
  bootstrap_servers = 'kafka:9092',
  topic = 'orders',
  format = 'json',
  'source.offset' = 'earliest',
  'source.read_mode' = 'read_committed'
);

CREATE TABLE paid_orders_sink WITH (
  connector = 'iceberg',
  type = 'sink',
  'catalog.type' = 'rest',
  'catalog.url' = 'http://lakekeeper:8181/catalog',
  'catalog.warehouse' = 'warehouse-id',
  namespace = 'analytics',
  table = 'paid_orders',
  append_only = 'true',
  infer_schema = 'true'
);

CREATE STREAMING PIPELINE orders_to_iceberg
WITH (
  parallelism = 4,
  checkpoint_interval = '10 seconds'
)
AS INSERT INTO paid_orders_sink
SELECT
  order_id,
  customer_id,
  amount,
  amount * 1.10 AS amount_with_tax
FROM orders_source
WHERE status = 'paid' AND amount > 0;
```

The first two results contain accepted table IDs. The pipeline result contains its pipeline ID, job ID, state, query, stop mode, and checkpoint interval. Workers start asynchronously after acceptance.

The sink takes its columns from the transformation because `infer_schema` is enabled. Its destination is `analytics.paid_orders` in the selected Iceberg warehouse.

## 2. Check that the pipeline is running

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
DESCRIBE PIPELINE orders_to_iceberg;
```

Read the JSON in the `result` column. Check that `state` reaches `Running`. The reported checkpoint interval is in microseconds: 10 seconds is `10000000`.

If the state reports a failure, inspect the pipeline through the existing pipeline controls. Use the shared [Observability](https://docs.e6data.com/transformation/operations/observability.md) docs for metrics and service visibility.

## 3. Send a small input

Use your Kafka producer to publish one message per line below to `orders`. Each message is a JSON object matching the source columns.

```text theme={"theme":{"light":"github-light","dark":"github-dark"}}
{"order_id":1,"customer_id":101,"amount":20,"status":"paid"}
{"order_id":2,"customer_id":102,"amount":30,"status":"pending"}
{"order_id":3,"customer_id":103,"amount":40,"status":"paid"}
{"order_id":4,"customer_id":104,"amount":-5,"status":"paid"}
```

With `read_committed`, a transactional producer must commit its transaction before those records can be read.

## 4. Verify the Iceberg output

Allow a successful checkpoint and sink publication to complete. Read the destination through your existing Iceberg query connection. If you need a Unified read connection, choose the matching [catalog recipe](https://docs.e6data.com/transformation/recipes/overview.md#catalog-connections) and connect it to the same warehouse.

Use the actual catalog name in this query. The text block below is a template to adapt after connecting the catalog.

```text theme={"theme":{"light":"github-light","dark":"github-dark"}}
SELECT order_id, customer_id, amount, amount_with_tax
FROM <your_catalog>.analytics.paid_orders
ORDER BY order_id;
```

For the four messages above, a fresh output table contains:

| order\_id | customer\_id | amount | amount\_with\_tax |
| - | - | - | - |
| 1 | 101 | 20 | 22 |
| 3 | 103 | 40 | 44 |

Order 2 is pending. Order 4 has a negative amount. Both fail the filter. Row arrival order is not guaranteed, so the read query sorts the result.

`paid_orders_sink` is a Streaming connector name. Use the qualified Iceberg table name when reading the output from Unified.

## 5. Stop the pipeline

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
STOP PIPELINE orders_to_iceberg;
DESCRIBE PIPELINE orders_to_iceberg;
```

`STOP` requests a checkpoint stop. Repeat `DESCRIBE` until `state` is `Stopped`. The immediately following `DESCRIBE` may still report a running or stopping state.

## Clean up

Stopping leaves the pipeline definition, connector definitions, and Iceberg output in place. Use the existing pipeline controls to delete the stopped pipeline. Remove the connector definitions after they have no pipeline users. Remove the test Iceberg table and Kafka topic separately through their owning systems when you no longer need the data.

The Streaming SQL commands do not yet include `DROP PIPELINE` or connector deletion. Running the create script again with new names creates another pipeline. An `earliest` source in a fresh pipeline can read the retained topic records again and append more rows.

## Continue

[Manage pipelines](https://docs.e6data.com/transformation/streaming/manage-pipelines) covers paused creation, resume, and request retries. [Transformations](https://docs.e6data.com/transformation/streaming/transformations) shows other ways to shape the same order records.


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