> For the complete documentation index, see [llms.txt](https://docs.e6data.com/ingestion-engine/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.e6data.com/ingestion-engine/get-started/first-pipeline.md). # 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](/ingestion-engine/connectors/lakehouse/iceberg.md). {% hint style="warning" %} 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. {% endhint %} ## Step 1: Create the HTTP Source Save this complete manifest as `events-source.yaml`: ```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: "" 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: ```bash 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`: ```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: "" warehouse: "s3:///warehouse" auth: token: "" namespace: analytics tableName: events appendOnly: true partitioning: fields: - name: event_time transform: hour shuffle_by_partition: enabled: true storageOptions: s3.endpoint: "" s3.region: "" s3.access-key-id: "" 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`: ```bash 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`: ```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: ```bash 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`: ```bash export HTTP_SOURCE_URL="://events./events" export HTTP_SOURCE_TOKEN="" ``` Send an event whose fields match the source schema: ```bash 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: ```bash kubectl get laminartables -n my-namespace kubectl get laminarpipelines -n my-namespace ``` For details about a resource that is not ready: ```bash 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: ```bash kubectl delete laminarpipeline events-pipeline -n my-namespace kubectl delete laminartable events-source events-sink -n my-namespace ``` ## Continue Continue with [Core Concepts](/ingestion-engine/get-started/core-concepts.md) to understand how profiles, tables, pipelines, and streaming execution fit together. --- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://docs.e6data.com/ingestion-engine/get-started/first-pipeline.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.