Your first pipeline

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.

Before you run: complete 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.

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

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

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

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

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 covers paused creation, resume, and request retries. Transformations shows other ways to shape the same order records.

Transformation Engine

Learn Laminar

Workloads

SQL reference

HTTP API

Recipes

Operations

1. Define the source, sink, and pipeline

2. Check that the pipeline is running

3. Send a small input

4. Verify the Iceberg output

5. Stop the pipeline

Clean up

Continue

​1. Define the source, sink, and pipeline

​2. Check that the pipeline is running

​3. Send a small input

​4. Verify the Iceberg output

​5. Stop the pipeline

​Clean up

​Continue

1. Define the source, sink, and pipeline

2. Check that the pipeline is running

3. Send a small input

4. Verify the Iceberg output

5. Stop the pipeline

Clean up

Continue