> 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/examples/kafka-to-iceberg.md). # 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 e6 Ingestion Engine deployment; * a Kafka cluster and a topic containing JSON events; * an Iceberg REST catalog; and * object storage accessible to the Iceberg catalog and e6 Ingestion Engine. Set the e6 Ingestion Engine API values used by every request: ```bash export INGESTION_ENGINE_URL="https://" export INGESTION_ENGINE_TOKEN="" ``` The example expects Kafka values with this shape: ```json { "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 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": ":9092,:9092", "authentication": { "type": "none" }, "connection_properties": {} } }' ``` Save the `id` from the response: ```bash export KAFKA_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](/ingestion-engine/connectors/message-queues/kafka.md). ## 2. Create the Kafka Source Table Create a source table for the `clickstream` topic: ```bash 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 export 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 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": "", "warehouse": "s3:///warehouse" }, "namespace": "analytics", "tableName": "clickstream", "appendOnly": true, "partitioning": { "fields": [ { "name": "event_time", "transform": "hour" } ], "shuffle_by_partition": { "enabled": true } }, "storageOptions": { "s3.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 export 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](/ingestion-engine/connectors/lakehouse/iceberg.md). 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 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 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 export 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 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](/ingestion-engine/get-started/examples/http-to-iceberg.md) for an API-driven pipeline that accepts events over HTTP. --- # 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/examples/kafka-to-iceberg.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.