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

# Kafka to Iceberg

> This example creates a streaming pipeline that reads JSON clickstream events from Kafka and writes them to an Apache Iceberg table.

You will create four public API resources:

* a reusable Kafka connection profile;
* a Kafka source table;
* an Iceberg sink table; and
* a SQL pipeline that connects the tables.

## Prerequisites

You need:

* the base URL and bearer token for a running Ingest Engine deployment;
* a Kafka cluster and a topic containing JSON events;
* an Iceberg REST catalog; and
* object storage accessible to the Iceberg catalog and Ingest Engine.

Set the Ingest Engine API values used by every request:

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
export INGESTION_ENGINE_URL="https://<your-ingest-engine-domain>"
export INGESTION_ENGINE_TOKEN="<your-api-token>"
```

The example expects Kafka values with this shape:

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "event_id": "evt-001",
  "user_id": 42,
  "session_id": "session-123",
  "page_url": "/products/123",
  "referrer": "/search",
  "event_type": "page_view",
  "event_time": "2026-07-15T10:30:00Z"
}
```

## 1. Create the Kafka Connection Profile

Create a profile containing the settings shared by Kafka tables:

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
curl --fail-with-body -sS -X POST \
  "$INGESTION_ENGINE_URL/api/v1/connectors/kafka/profiles" \
  -H "Authorization: Bearer $INGESTION_ENGINE_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "analytics_kafka",
    "config": {
      "bootstrap_servers": "<broker-1>:9092,<broker-2>:9092",
      "authentication": {
        "type": "none"
      },
      "connection_properties": {}
    }
  }'
```

Save the `id` from the response:

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
export KAFKA_PROFILE_ID="<profile-id>"
```

`bootstrap_servers` is a comma-separated broker list. This example uses an unauthenticated Kafka listener. For SASL or AWS MSK IAM, replace the `authentication` object with the configuration described in the [Kafka connector reference](/ingest/connectors/message-queues/kafka).

## 2. Create the Kafka Source Table

Create a source table for the `clickstream` topic:

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
curl --fail-with-body -sS -X POST \
  "$INGESTION_ENGINE_URL/api/v1/connectors/kafka/tables" \
  -H "Authorization: Bearer $INGESTION_ENGINE_TOKEN" \
  -H "Content-Type: application/json" \
  -d "{
    \"name\": \"clickstream\",
    \"connection_profile_id\": \"$KAFKA_PROFILE_ID\",
    \"config\": {
      \"topic\": \"clickstream\",
      \"type\": {
        \"source_config\": {
          \"offset\": \"earliest\",
          \"read_mode\": \"read_committed\"
        }
      }
    },
    \"schema\": {
      \"format\": {
        \"json\": {}
      },
      \"fields\": [
        {
          \"field_name\": \"event_id\",
          \"field_type\": {\"type\": {\"primitive\": \"String\"}},
          \"nullable\": false
        },
        {
          \"field_name\": \"user_id\",
          \"field_type\": {\"type\": {\"primitive\": \"Int64\"}},
          \"nullable\": false
        },
        {
          \"field_name\": \"session_id\",
          \"field_type\": {\"type\": {\"primitive\": \"String\"}},
          \"nullable\": false
        },
        {
          \"field_name\": \"page_url\",
          \"field_type\": {\"type\": {\"primitive\": \"String\"}},
          \"nullable\": false
        },
        {
          \"field_name\": \"referrer\",
          \"field_type\": {\"type\": {\"primitive\": \"String\"}},
          \"nullable\": true
        },
        {
          \"field_name\": \"event_type\",
          \"field_type\": {\"type\": {\"primitive\": \"String\"}},
          \"nullable\": false
        },
        {
          \"field_name\": \"event_time\",
          \"field_type\": {\"type\": {\"primitive\": \"DateTime\"}},
          \"nullable\": false
        }
      ]
    }
  }"
```

Save the returned table `id`:

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
export KAFKA_TABLE_ID="<kafka-table-id>"
```

The table `name` becomes its SQL name. `offset: earliest` starts a new source at the oldest retained offset, while `read_mode: read_committed` excludes records from aborted Kafka transactions.

## 3. Create the Iceberg Sink Table

Iceberg configuration is supplied directly on the table; it does not use a connection profile.

The following production payload uses a REST catalog, ambient object-store credentials, hourly partitioning, and explicit writer settings:

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
curl --fail-with-body -sS -X POST \
  "$INGESTION_ENGINE_URL/api/v1/connectors/iceberg/tables" \
  -H "Authorization: Bearer $INGESTION_ENGINE_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "clickstream_iceberg",
    "environment": "production",
    "config": {
      "catalog": {
        "provider": "direct",
        "format": "iceberg",
        "type": "rest",
        "url": "<iceberg-rest-catalog-url>",
        "warehouse": "s3://<bucket>/warehouse"
      },
      "namespace": "analytics",
      "tableName": "clickstream",
      "appendOnly": true,
      "partitioning": {
        "fields": [
          {
            "name": "event_time",
            "transform": "hour"
          }
        ],
        "shuffle_by_partition": {
          "enabled": true
        }
      },
      "storageOptions": {
        "s3.region": "<aws-region>"
      },
      "fileRotation": {
        "maxFileSizeBytes": 536870912
      },
      "parquet": {
        "compression": "zstd",
        "maxRowGroupRows": 1000000,
        "dataPageSize": 1048576,
        "dictionaryEnabled": true
      },
      "uploadChunkSizeBytes": 8388608,
      "uploadConcurrency": 8,
      "iceberg": {
        "targetFileSizeBytes": 536870912
      }
    },
    "schema": {
      "format": {
        "parquet": {}
      },
      "fields": [
        {
          "field_name": "event_id",
          "field_type": {"type": {"primitive": "String"}},
          "nullable": false
        },
        {
          "field_name": "user_id",
          "field_type": {"type": {"primitive": "Int64"}},
          "nullable": false
        },
        {
          "field_name": "session_id",
          "field_type": {"type": {"primitive": "String"}},
          "nullable": false
        },
        {
          "field_name": "page_url",
          "field_type": {"type": {"primitive": "String"}},
          "nullable": false
        },
        {
          "field_name": "referrer",
          "field_type": {"type": {"primitive": "String"}},
          "nullable": true
        },
        {
          "field_name": "event_type",
          "field_type": {"type": {"primitive": "String"}},
          "nullable": false
        },
        {
          "field_name": "event_time",
          "field_type": {"type": {"primitive": "DateTime"}},
          "nullable": false
        }
      ]
    }
  }'
```

Save the returned table `id`:

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
export ICEBERG_TABLE_ID="<iceberg-table-id>"
```

The sink writes to `analytics.clickstream` and partitions records by the hour of `event_time`. If the REST catalog requires authentication, add its `auth` object as described in the [Iceberg connector reference](/ingest/connectors/lakehouse/iceberg). Add only the storage options required by your object store and identity model.

## 4. Validate the SQL Query

Validate the query before creating the pipeline:

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
curl --fail-with-body -sS -X POST \
  "$INGESTION_ENGINE_URL/api/v1/pipelines/validate_query" \
  -H "Authorization: Bearer $INGESTION_ENGINE_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "query": "INSERT INTO clickstream_iceberg SELECT event_id, user_id, session_id, page_url, referrer, event_type, event_time FROM clickstream"
  }'
```

A valid query returns a `graph` and an empty `errors` array. Resolve every entry in `errors` before creating the pipeline.

## 5. Create the Pipeline

Create a production pipeline with four-way parallelism and a 10-second checkpoint interval:

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
curl --fail-with-body -sS -X POST \
  "$INGESTION_ENGINE_URL/api/v1/pipelines" \
  -H "Authorization: Bearer $INGESTION_ENGINE_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "kafka_to_iceberg_clickstream",
    "query": "INSERT INTO clickstream_iceberg SELECT event_id, user_id, session_id, page_url, referrer, event_type, event_time FROM clickstream",
    "parallelism": 4,
    "checkpointIntervalMicros": 10000000,
    "environment": "production"
  }'
```

Save the pipeline `id` from the response:

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
export PIPELINE_ID="<pipeline-id>"
```

`checkpointIntervalMicros` is required for production pipelines. Iceberg publishes committed data during successful checkpoints, so newly processed records may not be visible until a checkpoint completes.

## 6. Monitor the Resources

Read the resources through their public API endpoints:

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
curl --fail-with-body -sS \
  -H "Authorization: Bearer $INGESTION_ENGINE_TOKEN" \
  "$INGESTION_ENGINE_URL/api/v1/connection_profiles/$KAFKA_PROFILE_ID"

curl --fail-with-body -sS \
  -H "Authorization: Bearer $INGESTION_ENGINE_TOKEN" \
  "$INGESTION_ENGINE_URL/api/v1/connection_tables/$KAFKA_TABLE_ID"

curl --fail-with-body -sS \
  -H "Authorization: Bearer $INGESTION_ENGINE_TOKEN" \
  "$INGESTION_ENGINE_URL/api/v1/connection_tables/$ICEBERG_TABLE_ID"

curl --fail-with-body -sS \
  -H "Authorization: Bearer $INGESTION_ENGINE_TOKEN" \
  "$INGESTION_ENGINE_URL/api/v1/pipelines/$PIPELINE_ID"

curl --fail-with-body -sS \
  -H "Authorization: Bearer $INGESTION_ENGINE_TOKEN" \
  "$INGESTION_ENGINE_URL/api/v1/pipelines/$PIPELINE_ID/jobs"
```

The pipeline response includes its compiled graph and current action fields. The jobs response reports execution state and failure details. The table responses show the resolved connector, schema, configuration, and consumer count.

## Continue

Continue with [HTTP to Iceberg](/ingest/examples/http-to-iceberg) for an API-driven pipeline that accepts events over HTTP.


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