Schema Registry

How-to guideApplies to: All deployments

Connect Kafka or Confluent sources to Confluent Schema Registry or AWS Glue Schema Registry to read registered Avro events. Ingest Engine resolves each record’s schema automatically. For topics that carry events from several source tables, configure an Iceberg multi-table sink to route each table’s events to its own destination.

This pipeline stores the full event history. Database updates and deletes are kept as events; they aren’t applied as changes to the current rows in Iceberg.

Before you start

You need:

A reachable Kafka topic containing Avro records encoded with Confluent or AWS Glue Schema Registry.
Read access to the topic, its consumer group and every schema used by the topic.
An Iceberg catalog, warehouse and destination namespace, with permission to create and write tables.
A non-empty string in each record’s nested source.table field, naming the destination table.

For AWS MSK and Glue, use Cross Account Access to configure the source and registry roles. Broker connectivity, registry access and Iceberg storage access must all be available to your deployment.

Configure the pipeline in the console

Create a Kafka source

Enter the broker addresses, authentication, topic and starting offset, then enable Multi-schema source.

Select the schema registry provider

For Glue, enter Glue region, Registry name and, if required, Registry role ARN. For Confluent, enter Registry endpoint and the separate registry credentials. You don’t need to declare a fixed source schema.

Create an Iceberg sink

Enter the catalog, warehouse, namespace and storage settings, then enable Multi-table routing. Leave the fixed Iceberg table name and sink schema unset.

Create the pipeline

Review the generated pipeline query and create the pipeline. For a SQL workflow, the equivalent query is:

INSERT INTO multi_schema_iceberg
SELECT * FROM multi_schema_events;

Use the resource names you created. Keep this pipeline as a direct source-to-sink transfer.

Choose a schema registry

In API or YAML configuration, set config.multi_schema: true, choose schema_registry_provider, and supply schema_registry. Omit the source table’s schema. Registry-backed ingestion in this mode supports Avro. It doesn’t support JSON or Protobuf.

AWS Glue
Confluent

multi_schema: true
schema_registry_provider: glue
schema_registry:
  region: us-east-1
  registry_name: production
  role_arn: "<registry-iam-role-arn>"

Omit role_arn to use the deployment’s default AWS credentials. To reuse a configured MSK source role instead, remove the registry role ARN and set use_source_role: true. The reused role must have both MSK and Glue read permissions.

multi_schema: true
schema_registry_provider: confluent
schema_registry:
  endpoint: https://registry.example.com
  api_key: "${SCHEMA_REGISTRY_API_KEY}"
  api_secret: "${SCHEMA_REGISTRY_API_SECRET}"

Registry credentials are separate from broker credentials. Ingest Engine selects the registered schema associated with each record, so no subject override or fixed schema ID is needed.

How destination tables are selected

Records with source.table set to orders are written to the orders table in the configured Iceberg namespace. Records with source.table set to customers are written to customers. The topic name and registry schema name don’t set the destination name. New table names can arrive while the pipeline is running. The sink creates missing tables from the incoming data. Missing, empty, null or non-string source.table values fail ingestion, as do destination names the catalog rejects. The complete Avro event is kept, including the Debezium before, after, source, op and timestamp fields. Kafka tombstones are skipped. Two databases that use the same source.table value share a destination table, so use separate pipelines or namespaces if those histories must stay separate.

Configuration and limits

Set multiTable: {} on the Iceberg sink, omit tableName, and leave multiTable.tableColumn unset for this pipeline. Tables may have different schemas, but a new schema version for an existing destination must keep compatible column names, order and types. Incompatible changes fail the pipeline; automatic schema evolution isn’t supported in this mode.

Multi-table mode is append-only. It doesn’t support primary keys or appendOnly: false.
multiTable.maxTables defaults to 5,000 destination tables per sink task. Exceeding the limit fails processing.
Don’t enable inferSchema, sortColumns, indexes or evolution for this pipeline. The e6 committer-service catalog isn’t supported in multi-table mode.
If partitioning is configured, every destination’s data must contain the configured partition fields.
Data becomes visible through successful checkpoints. More active tables and partitions increase checkpoint work.
Kafka doesn’t guarantee record ordering across partitions.

See Multi-table routing on the Iceberg connector page.

Importable YAML

This example uses an unauthenticated development broker and the deployment’s AWS credentials for Glue and S3. Replace the broker, registry, catalog, warehouse, namespace and authentication settings before using it in your environment. For explicit AWS role configuration, see Cross Account Access.

# 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;

SQL example

Set the catalog warehouse, and the broker, registry and storage authentication options your deployment needs. The complete SQL source, sink and transfer are:

CREATE TABLE events WITH (
  connector = 'kafka', type = 'source',
  bootstrap_servers = 'localhost:9092', topic = 'events',
  'source.offset' = 'earliest', multi_schema = 'true',
  schema_registry_provider = 'glue',
  'schema_registry.region' = 'us-east-1',
  'schema_registry.registry_name' = 'events'
);

CREATE TABLE iceberg_events WITH (
  connector = 'iceberg', type = 'sink',
  'catalog.type' = 'rest', 'catalog.url' = 'http://lakekeeper:8181/catalog',
  namespace = 'analytics', append_only = 'true', 'multi_table.enabled' = 'true'
);

INSERT INTO iceberg_events SELECT * FROM events;

Ingest Engine

Get started

Connectors

Streaming SQL

Operate

Reference

Concepts

Before you start

Configure the pipeline in the console

Choose a schema registry

How destination tables are selected

Configuration and limits

Importable YAML

SQL example

​Before you start

​Configure the pipeline in the console

​Choose a schema registry

​How destination tables are selected

​Configuration and limits

​Importable YAML

​SQL example

Before you start

Configure the pipeline in the console

Choose a schema registry

How destination tables are selected

Configuration and limits

Importable YAML

SQL example