Skip to main content

Kafka to Iceberg Streaming Job

The Kafka to Iceberg Streaming Job copies records from Kafka topics into an Iceberg table in the IOMETE Lakehouse using Spark Structured Streaming. Point it at a Kafka cluster and a topic pattern, and it continuously (or on a trigger) appends new records to the destination table.

Deserialization Format​

Only JSON deserialization is currently supported.

Installation​

Marketplace​

Open Job Templates and click Marketplace. Find the kafka-iceberg-stream card, click the ⋮ menu, and select Deploy.

Deploy kafka-streaming from the Marketplace | IOMETEDeploy kafka-streaming from the Marketplace | IOMETE

The job form opens pre-filled with recommended defaults. Update the Kafka bootstrap servers, subscription pattern, and destination table in the config file (see Configuration Properties), then click Create.

Manual Setup​

Open Streaming Jobs in the sidebar and click New Streaming Job if you do not use the Marketplace flow.

Streaming Jobs page with New Streaming Job button | IOMETEStreaming Jobs page with New Streaming Job button | IOMETE

1. Name and Application

  • Name: any name you like, for example kafka-iceberg-stream
  • Application type: Python
  • Docker image: iomete.azurecr.io/iomete/kafka-iceberg-stream:1.2.0 (replace with the latest version)
  • Main application file: local:///app/job.py
  • Java options (optional): -Dlog4j.configurationFile=/opt/spark/iomete/log4j2.properties, specify a logging configuration file
Application type, Docker image, and main application file fields | IOMETEApplication type, Docker image, and main application file fields | IOMETE

2. Config File

Expand Configurations, select the Config Maps tab, click Add config, and paste the HOCON template below. See Configuration Properties for what each field does.

Config Maps tab with the HOCON configuration file | IOMETEConfig Maps tab with the HOCON configuration file | IOMETE
{
  kafka: {
    options: {
      "kafka.bootstrap.servers": "kafka-bootstrap-server:9092",
      "subscribePattern": ".*"
    },

    # either once or processing_time should be set for the trigger. Not both.
    trigger: {
      # processing_time: "15 minutes",
      once: True
    },

    # set checkpoint_location to object storage for production.
    # Example: "s3a://assets-dir/checkpoints/kafka-streaming/data/app1"
    checkpoint_location: ".tmp/checkpoints/kafka-streaming/data/app1",
  },
  destination: {
    database: "default",
    table: "all_db_changes_v1",
  }
}
note

It's recommended to exclude the startingOffsets option. If the destination table doesn't exist or is empty, the job defaults to earliest automatically. If the table already has data, it defaults to latest. This lets the job start from the beginning on the first run and continue from where it left off (based on the checkpoint state) on every run after that.

3. Instance Resources

Pick driver and executor instance types that fit your throughput.

Deployment type, node driver, and executor fields | IOMETEDeployment type, node driver, and executor fields | IOMETE

Click Create to save the job.

Configuration Properties​

FieldDescription
kafka.optionsKafka consumer options, passed through as-is. See Kafka Consumer Configurations for the full list.
kafka.triggerTrigger options. See Triggers for details. Only one of processing_time or once may be set.
kafka.checkpoint_locationCheckpoint location. See Checkpointing for details. Set this to object storage for production, for example s3a://assets-dir/checkpoints/kafka-streaming/data/app1.
destination.databaseIceberg database name. The database must already exist.
destination.tableIceberg table name. The job creates the table if it doesn't exist.

The job creates the destination table with the following schema:

root
 |-- topic: string (nullable = true)
 |-- partition: integer (nullable = true)
 |-- offset: long (nullable = true)
 |-- timestamp: timestamp (nullable = true)
 |-- timestampType: integer (nullable = true)
 |-- key: string (nullable = true)
 |-- value: string (nullable = true)
ColumnDescription
topicThe Kafka topic the record was received from.
partitionThe Kafka partition the record was received from.
offsetThe Kafka offset of the record.
timestampThe Kafka timestamp of the record.
timestampTypeThe Kafka timestamp type of the record.
keyThe Kafka record key.
valueThe Kafka record value.

Running the Job​

With trigger.once set, the job processes everything currently available on the topic and stops. Trigger it again (or run it on a schedule) to pick up new records. With trigger.processing_time set, the job runs continuously, checking for new records at the given interval.

To trigger a run on demand, open the job and click Run.

ON THIS PAGE