Build your first pipeline

You will create:

an HTTP source table;
an Iceberg sink table; and
a SQL pipeline that connects them.

Prerequisites

You need:

access to a running Ingest Engine cluster;
the Kubernetes namespace and name of that cluster;
an Iceberg REST catalog;
an S3 or S3-compatible object store; and
kubectl access to the cluster.

The example uses an Iceberg REST catalog and S3-compatible storage. For other catalog and storage combinations, see Apache Iceberg.

Replace my-namespace, my-cluster, and every value enclosed in <...> before applying the manifests. Kubernetes does not substitute shell variables in YAML files. Do not commit real tokens or storage credentials to source control.

Step 1: Create the HTTP Source

Save this complete manifest as events-source.yaml:

apiVersion: laminar.stream/v1alpha1
kind: LaminarTable
metadata:
  name: events-source
  namespace: my-namespace
spec:
  clusterRef: my-cluster
  connector: http_source
  route: events
  config:
    name: events_source
    config:
      port: 8080
      path: /events
      bind_address: 0.0.0.0
      buffer_size: 1000
      max_body_size: 5242880
      auth:
        type: bearer
        token: "<http-source-token>"
    schema:
      format:
        json: {}
      framing:
        method:
          newline:
            maxLineLength: 1048576
      bad_data:
        fail: {}
      fields:
        - field_name: event_id
          field_type:
            type:
              primitive: String
          nullable: false
        - field_name: user_id
          field_type:
            type:
              primitive: String
          nullable: false
        - field_name: event_type
          field_type:
            type:
              primitive: String
          nullable: false
        - field_name: event_time
          field_type:
            type:
              primitive: DateTime
          nullable: false
        - field_name: payload
          field_type:
            type:
              primitive: String
          nullable: true

Apply it:

kubectl apply -f events-source.yaml
kubectl get laminartable events-source -n my-namespace

The important source settings are:

Setting	Purpose
`route`	Selects the external hostname configured by the cluster gateway. It must be a lowercase DNS label.
`port`	Port on which the source accepts requests.
`path`	HTTP path that accepts `POST` requests.
`buffer_size`	Maximum number of requests that can wait for processing.
`max_body_size`	Maximum decompressed request size in bytes.
`auth`	Requires clients to authenticate with a bearer token. Basic authentication is also supported.
`framing.method.newline`	Allows one JSON record or multiple newline-delimited JSON records per request.
`bad_data.fail`	Stops processing when an input record does not match the schema. Use `drop: {}` only when discarding malformed records is acceptable.

The source schema and format are required. Production clusters also require explicit capacity, authentication, and bad-data settings, all of which are included above.

Step 2: Create the Iceberg Sink

Save this complete manifest as events-sink.yaml:

apiVersion: laminar.stream/v1alpha1
kind: LaminarTable
metadata:
  name: events-sink
  namespace: my-namespace
spec:
  clusterRef: my-cluster
  connector: iceberg
  maintenance:
    enabled: false
  config:
    name: events_sink
    config:
      catalog:
        provider: direct
        format: iceberg
        type: rest
        url: "<iceberg-rest-catalog-url>"
        warehouse: "s3://<bucket>/warehouse"
        auth:
          token: "<iceberg-catalog-token>"
      namespace: analytics
      tableName: events
      appendOnly: true
      partitioning:
        fields:
          - name: event_time
            transform: hour
        shuffle_by_partition:
          enabled: true
      storageOptions:
        s3.endpoint: "<s3-endpoint>"
        s3.region: "<s3-region>"
        s3.access-key-id: "<s3-access-key-id>"
        s3.secret-access-key: "<s3-secret-access-key>"
        s3.path-style-access: "true"
        s3.disable-config-load: "true"
      fileRotation:
        maxFileSizeBytes: 134217728
      parquet:
        compression: zstd
        compressionLevel: 3
        maxRowGroupRows: 1000000
        dataPageSize: 1048576
        dictionaryEnabled: true
        dictionaryPageSize: 1048576
      uploadChunkSizeBytes: 8388608
      uploadConcurrency: 8
      iceberg:
        targetFileSizeBytes: 268435456
    schema:
      format:
        parquet: {}
      fields:
        - field_name: event_id
          field_type:
            type:
              primitive: String
          nullable: false
        - field_name: user_id
          field_type:
            type:
              primitive: String
          nullable: false
        - field_name: event_type
          field_type:
            type:
              primitive: String
          nullable: false
        - field_name: event_time
          field_type:
            type:
              primitive: DateTime
          nullable: false
        - field_name: payload
          field_type:
            type:
              primitive: String
          nullable: true

If your REST catalog is anonymous, remove the auth block. When using AWS S3 with workload identity or the default credential chain, remove the static access keys and any custom endpoint settings that are not needed. Apply the sink and wait until both tables report Ready:

kubectl apply -f events-sink.yaml
kubectl wait --for=jsonpath='{.status.phase}'=Ready \
  laminartable/events-source laminartable/events-sink \
  -n my-namespace \
  --timeout=5m

The important sink settings are:

Setting	Purpose
`connector: iceberg`	Selects the current Apache Iceberg sink.
`catalog.provider: direct`	Connects the sink directly to the configured catalog.
`catalog.format: iceberg`	Selects the Iceberg catalog protocol.
`catalog.type: rest`	Uses an Iceberg REST catalog.
`namespace` and `tableName`	Identify the destination table as `analytics.events`.
`appendOnly: true`	Accepts insert-only output. CDC output requires primary keys and additional constraints.
`partitioning`	Partitions the table by hour using `event_time`.
`storageOptions`	Configures access to the object store used by the table.
`fileRotation` and `parquet`	Control output file sizing and Parquet encoding.
`uploadChunkSizeBytes` and `uploadConcurrency`	Control multipart upload behavior.
`iceberg.targetFileSizeBytes`	Sets the target data-file size used by Iceberg maintenance.
`maintenance.enabled`	Makes the maintenance policy explicit. This first pipeline disables scheduled maintenance.

The sink schema must match the columns and types produced by the pipeline query.

Step 3: Create the Pipeline

Save this complete manifest as events-pipeline.yaml:

apiVersion: laminar.stream/v1alpha1
kind: LaminarPipeline
metadata:
  name: events-pipeline
  namespace: my-namespace
spec:
  clusterRef: my-cluster
  replicas: 1
  config:
    name: events_pipeline
    query: |
      INSERT INTO events_sink
      SELECT
        event_id,
        user_id,
        event_type,
        event_time,
        payload
      FROM events_source
    parallelism: 1
    checkpointIntervalMicros: 10000000

The SQL names are the config.name values from the two table manifests: events_source and events_sink. Apply the pipeline:

kubectl apply -f events-pipeline.yaml
kubectl wait --for=jsonpath='{.status.phase}'=Running \
  laminarpipeline/events-pipeline \
  -n my-namespace \
  --timeout=5m

parallelism: 1 is a safe starting point for this HTTP ingestion pipeline. The checkpoint interval is expressed in microseconds, so 10000000 means 10 seconds.

Step 4: Send an Event

Build the public source URL from the hostname assigned to route: events and the configured path: /events:

export HTTP_SOURCE_URL="<scheme>://events.<gateway-domain>/events"
export HTTP_SOURCE_TOKEN="<http-source-token>"

Send an event whose fields match the source schema:

curl --fail-with-body -X POST "$HTTP_SOURCE_URL" \
  -H "Authorization: Bearer $HTTP_SOURCE_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "event_id": "evt-001",
    "user_id": "user-42",
    "event_type": "click",
    "event_time": "2026-07-15T10:30:00Z",
    "payload": "{\"page\":\"/home\"}"
  }'

A successful request returns HTTP 200, which means the event was accepted for processing. It does not confirm that a downstream checkpoint has completed. Iceberg commits occur during checkpoints, so table visibility can lag behind the HTTP response.

Step 5: Verify the Pipeline

Check the resource phases:

kubectl get laminartables -n my-namespace
kubectl get laminarpipelines -n my-namespace

For details about a resource that is not ready:

kubectl describe laminartable events-source -n my-namespace
kubectl describe laminartable events-sink -n my-namespace
kubectl describe laminarpipeline events-pipeline -n my-namespace

Common HTTP responses are:

Response	Meaning
`200`	The request was accepted.
`400`	The request body is empty.
`401`	The bearer token is missing or incorrect.
`404`	The request path does not match `config.path`.
`413`	The request exceeds `max_body_size`.
`429`	The source buffer is full; retry with backoff.
`503`	The source is temporarily unavailable; retry.

Clean Up

Delete the pipeline before deleting its tables:

kubectl delete laminarpipeline events-pipeline -n my-namespace
kubectl delete laminartable events-source events-sink -n my-namespace

Continue

Continue with Core Concepts to understand how profiles, tables, pipelines, and streaming execution fit together.

Keep going across e6data

Query the tables you just wrote

Operate your pipelines

Configuration, observability and metrics for running pipelines.

Ingest Engine

Get started

Connectors

Streaming SQL

Operate

Reference

Concepts

Build your first pipeline

Prerequisites

Step 1: Create the HTTP Source

Step 2: Create the Iceberg Sink

Step 3: Create the Pipeline

Step 4: Send an Event

Step 5: Verify the Pipeline

Clean Up

Continue

Keep going across e6data

Query the tables you just wrote

Operate your pipelines

​Prerequisites

​Step 1: Create the HTTP Source

​Step 2: Create the Iceberg Sink

​Step 3: Create the Pipeline

​Step 4: Send an Event

​Step 5: Verify the Pipeline

​Clean Up

​Continue

​Keep going across e6data

Query the tables you just wrote

Operate your pipelines

Prerequisites

Step 1: Create the HTTP Source

Step 2: Create the Iceberg Sink

Step 3: Create the Pipeline

Step 4: Send an Event

Step 5: Verify the Pipeline

Clean Up

Continue

Keep going across e6data