Kafka to Iceberg

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:

export INGESTION_ENGINE_URL="https://<your-ingest-engine-domain>"
export INGESTION_ENGINE_TOKEN="<your-api-token>"

The example expects Kafka values with this shape:

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

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:

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.

2. Create the Kafka Source Table

Create a source table for the clickstream topic:

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:

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:

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:

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

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:

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:

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:

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 for an API-driven pipeline that accepts events over HTTP.

Ingest Engine

Get started

Connectors

Streaming SQL

Operate

Reference

Concepts

Prerequisites

1. Create the Kafka Connection Profile

2. Create the Kafka Source Table

3. Create the Iceberg Sink Table

4. Validate the SQL Query

5. Create the Pipeline

6. Monitor the Resources

Continue

​Prerequisites

​1. Create the Kafka Connection Profile

​2. Create the Kafka Source Table

​3. Create the Iceberg Sink Table

​4. Validate the SQL Query

​5. Create the Pipeline

​6. Monitor the Resources

​Continue

Prerequisites

1. Create the Kafka Connection Profile

2. Create the Kafka Source Table

3. Create the Iceberg Sink Table

4. Validate the SQL Query

5. Create the Pipeline

6. Monitor the Resources

Continue