Schema Registry to Iceberg (YAML)

How-to guideApplies to: All deployments

Use these examples when importing a pipeline as YAML. For console setup, see Schema Registry. Each example includes the source, destination and pipeline. Replace the connection settings for your environment before importing. Your deployment needs access to Kafka, Glue, the Iceberg catalog and its storage. For AWS roles and permissions, see Cross Account Access.

Write to one Iceberg table

This example reads Debezium Avro events from MSK using AWS Glue Schema Registry and writes them to one Iceberg table. Replace every <placeholder>. The source role needs both MSK and Glue read permissions. Configure the REST catalog’s warehouse for your S3 bucket and prefix, and create the destination namespace before starting the pipeline. If the catalog requires authentication, add its authentication settings.

---
apiVersion: laminar.io/v1
kind: Table
spec:
  name: cdc_source
  connector: kafka
  config:
    multi_schema: false
    authentication:
      type: aws_msk_iam
      aws_msk_iam_config:
        region: "<msk-region>"
        role_arn: "arn:aws:iam::<source-account-id>:role/<msk-glue-role-name>"
    bootstrap_servers: "<msk-iam-bootstrap-servers>"
    client_configs:
      group.id: "<consumer-group-id>"
    schema_registry_provider: glue
    schema_registry:
      region: "<glue-region>"
      registry_name: "<glue-registry-name>"
      use_source_role: true
    topic: "<kafka-topic>"
    type:
      source_config:
        group_id: "<consumer-group-id>"
        offset: latest
    value_subject: "<glue-schema-name>"
  schema:
    format:
      avro:
        flattenDebezium: true
    bad_data:
      fail: {}
---
apiVersion: laminar.io/v1
kind: Table
spec:
  name: cdc_iceberg_sink
  connector: iceberg
  config:
    catalog:
      provider: direct
      format: iceberg
      type: rest
      url: "<iceberg-rest-catalog-url>"
      warehouse: "<iceberg-warehouse>"
    namespace: "<iceberg-namespace>"
    tableName: "<iceberg-table-name>"
    appendOnly: true
    inferSchema: true
    storageOptions:
      s3.region: "<s3-bucket-region>"
      client.assume-role.arn: "arn:aws:iam::<s3-account-id>:role/<iceberg-writer-role-name>"
---
apiVersion: laminar.io/v1
kind: Pipeline
spec:
  name: cdc_to_iceberg
  query: |-
    INSERT INTO cdc_iceberg_sink
    SELECT * FROM cdc_source;
  parallelism: 1
  checkpointIntervalMicros: 5000000

The S3 writer role is separate from the source role. Use the bucket’s region, and grant the writer role access to the warehouse bucket and prefix. If it requires an external ID, add client.assume-role.external-id under storageOptions. If your deployment or catalog already supplies storage credentials, omit client.assume-role.arn. See S3 access. Start the source before publishing new events: this example reads only records arriving after it starts. Wait for a successful checkpoint, then query the destination table. Each event is stored as a new row, including updates and deletes; the pipeline does not apply changes to existing rows.

Write to several Iceberg tables

Use this example when one topic contains events from several source tables. Each event must have a non-empty string in source.table, which names its destination table. Missing destination tables are created automatically. This sample uses an unauthenticated development broker and the deployment’s AWS credentials for Glue and S3. Replace the broker, authentication, registry, catalog, warehouse, namespace and storage settings for your environment.

# Replace the Kafka, Glue, catalog and warehouse settings for your environment.
apiVersion: laminar.io/v1
kind: Table
spec:
  name: multi_schema_events
  connector: kafka
  config:
    bootstrap_servers: kafka:9092
    authentication:
      type: none
    topic: cdc-events
    type:
      source_config:
        offset: earliest
    multi_schema: true
    schema_registry_provider: glue
    schema_registry:
      region: us-east-1
      registry_name: cdc-events
---
apiVersion: laminar.io/v1
kind: Table
spec:
  name: multi_schema_iceberg
  connector: iceberg
  config:
    catalog:
      type: rest
      url: http://lakekeeper:8181/catalog
      warehouse: cdc-warehouse
    namespace: analytics
    appendOnly: true
    multiTable: {}
    storageOptions:
      s3.region: us-east-1
---
apiVersion: laminar.io/v1
kind: Pipeline
spec:
  name: multi_schema_kafka_to_iceberg
  parallelism: 1
  checkpointIntervalMicros: 3000000
  query: |
    INSERT INTO multi_schema_iceberg
    SELECT * FROM multi_schema_events;

This pipeline stores complete events as new rows. Use separate pipelines or namespaces if different databases use the same table names and their data must stay separate. Review multi-table routing limits, including supported schema changes, before starting. Start the pipeline, wait for a successful checkpoint, and query each destination table to verify that the events arrived.

Ingest Engine

Get started

Set up

Connectors

Streaming SQL

Operate

Reference

Concepts

Schema Registry to Iceberg (YAML)

Write to one Iceberg table

Write to several Iceberg tables

​Write to one Iceberg table

​Write to several Iceberg tables

Write to one Iceberg table

Write to several Iceberg tables