Kafka connect iceberg sink

​

Prerequisites

• Setup the Iceberg Catalog
• Create the Iceberg connector control topic, which cannot be used by other connectors.

​

Quick Start

1. Setup the kcctl client: doc
2. Set up the Iceberg Catalog, we can use the below yaml file to create a local iceberg catalog in k8s:
   Copy

   Ask AI

   apiVersion: v1
   data:
     spark-defaults.conf: |
       #
       # Licensed to the Apache Software Foundation (ASF) under one or more
       # contributor license agreements.  See the NOTICE file distributed with
       # this work for additional information regarding copyright ownership.
       # The ASF licenses this file to You under the Apache License, Version 2.0
       # (the "License"); you may not use this file except in compliance with
       # the License.  You may obtain a copy of the License at
       #
       #    http://www.apache.org/licenses/LICENSE-2.0
       #
       # Unless required by applicable law or agreed to in writing, software
       # distributed under the License is distributed on an "AS IS" BASIS,
       # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
       # See the License for the specific language governing permissions and
       # limitations under the License.
       #
   
       # Default system properties included when running spark-submit.
       # This is useful for setting default environmental settings.
   
       # Example:
       spark.sql.extensions                   org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
       spark.sql.catalog.demo                 org.apache.iceberg.spark.SparkCatalog
       spark.sql.catalog.demo.type            rest
       spark.sql.catalog.demo.uri             http://iceberg-rest.default.svc.cluster.local:8181
       spark.sql.catalog.demo.io-impl         org.apache.iceberg.aws.s3.S3FileIO
       spark.sql.catalog.demo.warehouse       s3://warehouse/
       spark.sql.catalog.demo.s3.endpoint     http://minio.default.svc.cluster.local:9000
       spark.sql.defaultCatalog               demo
       spark.eventLog.enabled                 true
       spark.eventLog.dir                     /home/iceberg/spark-events
       spark.history.fs.logDirectory          /home/iceberg/spark-events
       spark.sql.catalogImplementation        in-memory
   kind: ConfigMap
   metadata:
     name: spark-config
     namespace: default
   ---
   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: spark-iceberg
     namespace: default
   spec:
     replicas: 1
     selector:
       matchLabels:
         app: spark-iceberg
     template:
       metadata:
         labels:
           app: spark-iceberg
       spec:
         containers:
         - name: spark-iceberg
           image: tabulario/spark-iceberg
           ports:
             - name: one
               containerPort: 8888
             - name: two
               containerPort: 8080
             - name: three
               containerPort: 10000
             - name: four
               containerPort: 10001
           env:
             - name: AWS_ACCESS_KEY_ID
               value: admin
             - name: AWS_SECRET_ACCESS_KEY
               value: password
             - name: AWS_REGION
               value: us-east-1
           volumeMounts:
             - name: spark-config
               mountPath: /opt/spark/conf/spark-defaults.conf
               subPath: spark-defaults.conf
         volumes:
           - name: spark-config
             configMap:
               name: spark-config
   ---
   apiVersion: v1
   kind: Service
   metadata:
     name: spark-iceberg
     namespace: default
   spec:
     selector:
       app: spark-iceberg
     ports:
       - protocol: TCP
         name: one
         port: 8888
         targetPort: 8888
       - protocol: TCP
         name: two
         port: 8080
         targetPort: 8080
       - protocol: TCP
         name: three
         port: 10000
         targetPort: 10000
       - protocol: TCP
         port: 10001
         name: four
         targetPort: 10001
   ---
   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: iceberg-rest
     namespace: default
   spec:
     replicas: 1
     selector:
       matchLabels:
         app: iceberg-rest
     template:
       metadata:
         labels:
           app: iceberg-rest
       spec:
         containers:
         - name: iceberg-rest
           image: tabulario/iceberg-rest
           ports:
             - name: one
               containerPort: 8181
           env:
             - name: AWS_ACCESS_KEY_ID
               value: admin
             - name: AWS_SECRET_ACCESS_KEY
               value: password
             - name: AWS_REGION
               value: us-east-1
             - name: CATALOG_WAREHOUSE
               value: s3://warehouse/
             - name: CATALOG_IO__IMPL
               value: org.apache.iceberg.aws.s3.S3FileIO
             - name: CATALOG_S3_ENDPOINT
               value: http://minio.default.svc.cluster.local:9000
   ---
   apiVersion: v1
   kind: Service
   metadata:
     name: iceberg-rest
     namespace: default
   spec:
     selector:
       app: iceberg-rest
     ports:
       - protocol: TCP
         name: one
         port: 8181
         targetPort: 8181
   ---
   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: minio
     namespace: default
   spec:
     replicas: 1
     selector:
       matchLabels:
         app: minio
     template:
       metadata:
         labels:
           app: minio
       spec:
         hostname: warehouse
         subdomain: minio
         containers:
         - name: minio
           image: minio/minio
           args:
             - server
             - /data
             - --console-address
             - ":9001"
           ports:
             - name: one
               containerPort: 9000
             - name: two
               containerPort: 9001
           env:
             - name: MINIO_ROOT_USER
               value: admin
             - name: MINIO_ROOT_PASSWORD
               value: password
             - name: MINIO_DOMAIN
               value: minio.default.svc.cluster.local
   ---
   apiVersion: v1
   kind: Service
   metadata:
     name: minio
     namespace: default
   spec:
     selector:
       app: minio
     ports:
       - protocol: TCP
         name: one
         port: 9000
         targetPort: 9000
       - protocol: TCP
         name: two
         port: 9001
         targetPort: 9001
   

3. Initialize the Iceberg table:
   Copy

   Ask AI

   kubectl apply -f iceberg-spark.yaml
   kubectl wait -l app=spark-iceberg --for=condition=Ready pod --timeout=5m
   kubectl wait -l app=iceberg-rest --for=condition=Ready pod --timeout=5m
   kubectl wait -l app=minio --for=condition=Ready pod --timeout=5m
   
   sleep 30
   
   # initialize the bucket
   minio_pod_name=$(kubectl get pods -l app=minio -o=jsonpath='{.items[0].metadata.name}')
   kubectl exec $minio_pod_name -- /usr/bin/mc config host add minio http://minio.default.svc.cluster.local:9000 admin password
   kubectl exec $minio_pod_name -- /usr/bin/mc rm -r --force minio/warehouse || true
   kubectl exec $minio_pod_name -- /usr/bin/mc mb minio/warehouse
   kubectl exec $minio_pod_name -- /usr/bin/mc policy set public minio/warehouse
   

4. Create a JSON file like the following:
   Copy

   Ask AI

   {
       "name": "iceberg-sink",
       "config": {
           "connector.class": "io.tabular.iceberg.connect.IcebergSinkConnector",
           "topics": "kafka-iceberg-input",
           "iceberg.tables": "sink.kafka",
           "iceberg.catalog": "demo",
           "iceberg.catalog.type": "rest",
           "iceberg.catalog.uri": "http://iceberg-rest.default.svc.cluster.local:8181",
           "iceberg.catalog.client.region": "us-east-1",
           "iceberg.catalog.io-impl": "org.apache.iceberg.aws.s3.S3FileIO",
           "iceberg.catalog.warehouse": "s3://warehouse",
           "iceberg.catalog.s3.endpoint": "http://minio.default.svc.cluster.local:9000",
           "iceberg.catalog.s3.path-style-access": "true",
           "iceberg.catalog.s3.access-key-id": "admin",
           "iceberg.catalog.s3.secret-access-key": "password",
           "iceberg.tables.auto-create-enabled": "true",
           "iceberg.tables.evolve-schema-enabled": "true",
           "iceberg.control.commit.interval-ms": "1000",
           "key.converter": "org.apache.kafka.connect.storage.StringConverter",
           "value.converter": "org.apache.kafka.connect.json.JsonConverter",
           "value.converter.schemas.enable": "false",
           "tasks.max": "1"
       }
   }
   

5. Run the following command to create the connector:
   Copy

   Ask AI

   kcctl create -f <filename>.json
   

​

Quick Start 2 - Write to AWS S3 Table

1. Create an AWS S3 table bucket, this is a new bucket type, A regular S3 bucket won’t work for S3 Tables, here we use kc-test-iceberg-table as an example.
2. Create a new database in the table bucket, you can go to the AWS Lake Formation console, and then go to the Data Catalog/Databases section to create it, here we use my_s3_namespace as an example.
3. Create an AWS IAM user and attach the following policy to it, replace eu-north-1, account-id, kc-test-iceberg-table and my_s3_namespace with your real region, account-id, table-bucket and database respectively.:
   Copy

   Ask AI

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "VisualEditor0",
               "Effect": "Allow",
               "Action": "lakeformation:GetDataAccess",
               "Resource": "*"
           },
           {
               "Sid": "GlueAndCatalogForS3Tables",
               "Effect": "Allow",
               "Action": [
                   "glue:GetCatalog",
                   "glue:GetDatabase",
                   "glue:GetDatabases",
                   "glue:GetTable",
                   "glue:GetTables",
                   "glue:CreateTable",
                   "glue:UpdateTable"
               ],
               "Resource": [
                   "arn:aws:glue:eu-north-1:{account_id}:catalog",
                   "arn:aws:glue:eu-north-1:{account_id}:catalog/s3tablescatalog",
                   "arn:aws:glue:eu-north-1:{account_id}:catalog/s3tablescatalog/kc-test-iceberg-table",
                   "arn:aws:glue:eu-north-1:{account_id}:database/s3tablescatalog/kc-test-iceberg-table/my_s3_namespace",
                   "arn:aws:glue:eu-north-1:{account_id}:table/s3tablescatalog/kc-test-iceberg-table/my_s3_namespace/*"
               ]
           }
       ]
   }
   

4. Create an access key for the IAM user, you will need the Access Key ID and Secret Access Key later.
5. In the Lake Formation console, grant the user database permissions (Create table, Describe) on [account-id]:s3tablescatalog/kc-test-iceberg-table/my_s3_namespace, and table permissions (Super on ALL_TABLES or at least the table you’ll write).
6. Create a JSON file like the following, replace the fields such as region, account-id, table_bucket, access_key_id and secret_access_key with your real values:
   Copy

   Ask AI

   {
       "name": "test-ice",
       "config": {
           "name": "test-ice",
           "connector.class": "io.tabular.iceberg.connect.IcebergSinkConnector",
           "topics": "events",
           "tasks.max": "1",
           "iceberg.tables.auto-create-enabled": "true",
           "iceberg.tables": "my_s3_namespace.events",
           "iceberg.tables.evolve-schema-enabled": "true",
           "header.converter": "org.apache.kafka.connect.storage.SimpleHeaderConverter",
           "value.converter": "org.apache.kafka.connect.json.JsonConverter",
           "value.converter.schemas.enable": "false",
           "key.converter": "org.apache.kafka.connect.storage.StringConverter",
           "key.converter.schemas.enable": "false",
           "iceberg.catalog": "iceberg",
           "iceberg.catalog.type": "rest",
           "iceberg.catalog.uri": "https://glue.{region}.amazonaws.com/iceberg",
           "iceberg.catalog.warehouse": "{account-id}:s3tablescatalog/{table_bucket}",
           "iceberg.catalog.rest.access-key-id": "{access_key_id}",
           "iceberg.catalog.rest.secret-access-key": "{secret_access_key}",
           "iceberg.catalog.rest.sigv4-enabled": "true",
           "iceberg.catalog.rest.signing-name": "glue",
           "iceberg.catalog.rest.signing-region": "{region}",
           "iceberg.control.topic": "control-iceberg",
           "sn.connector.image.opts": "-Daws.region={region}"
       }
   }
   

7. Run the following command to create the connector:
   Copy

   Ask AI

   kcctl create -f <filename>.json
   

8. Produce some test data to the events topic, you can use the following command to produce some test data:
   Copy

   Ask AI

   echo '{"id": 1, "name": "Jack", "message": "hello iceberg", "email": "Jack@test.com"}' | ~/kafka/kafka_2.13-3.1.0/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic events
   

9. Wait until the data is written to the Iceberg table, you can check the logs of the connector to see something like below:
   Copy

   Ask AI

   2025-08-25T08:05:27,622+0000 [iceberg-coord] INFO  io.tabular.iceberg.connect.channel.Coordinator - Commit fa40ffe5-9424-4994-8439-cebe46919ff4 complete, committed to 1 table(s), vtts null
   

10. Check the Iceberg table in the AWS Athena console, you should see the events table in the my_s3_namespace database, and you can run a query like below to see the data:
    Copy

    Ask AI

    SELECT * FROM "my_s3_namespace"."events" limit 10;
    

​

Limitations

• Each Iceberg sink connector must have its own control topic.

​

Configuration