Build your first pipeline

This guide builds an end-to-end pipeline that accepts JSON events over HTTP and writes them to an Apache Iceberg table.

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