Iceberg Sink

The Iceberg Sink Connector allows you to consume messages from Iggy topics and store them in Iceberg tables.

Features

• Support for S3-compatible storage
• Support for REST catalogs
• Single destination table
• Multiple-table fan-out static routing
• Multiple-table fan-out dynamic routing

Configuration example

[plugin_config]
tables = ["nyc.users"]
catalog_type = "rest"
warehouse = "warehouse"
uri = "http://localhost:8181"
dynamic_routing = true
dynamic_route_field = "db_table"
store_url = "http://localhost:9000"
store_access_key_id = "admin"
store_secret_access_key = "password"
store_region = "us-east-1"
store_class = "s3"

Configuration Options

• tables: The names of the Iceberg tables you want to statically route Iggy messages to. The name should include the table’s namespace, separated by a dot (.).
• catalog_type: The type of catalog you are routing data to. Currently, only REST catalogs are fully supported.
• warehouse: The name of the bucket or warehouse where Iggy will upload data files.
• uri: The URI of the Iceberg catalog.
• dynamic_routing: Enables dynamic routing. See more details later in this document.
• dynamic_route_field: The name of the message field that specifies the Iceberg table to route data to. See more details below.
• store_url: The URL of the object storage for data uploads.
• store_access_key_id: The access key ID of the object storage.
• store_secret_access_key: The secret key used to upload data to the object storage.
• store_region: The region of the object storage. Required. For S3-compatible stores that ignore it, supply any placeholder value.
• store_class: The storage class to use. Currently, only S3-compatible storage is supported.

Dynamic Routing

If you don't know the names of the Iceberg tables you want to route data to in advance, you can use the dynamic routing feature. Insert a field in your Iggy messages with the name of the Iceberg table the message should be routed to. The Iggy connector will parse this field at runtime and route the message to the correct table.

The Iggy Iceberg Connector will skip messages in the following cases:

• The table declared in the message field does not exist.
• The message does not contain the field specified in the dynamic_route_field configuration option.

Dynamic routing configuration example

[plugin_config]
tables = [""]
catalog_type = "rest"
warehouse = "warehouse"
uri = "http://localhost:8181"
dynamic_routing = true
dynamic_route_field = "db_table"
store_url = "http://localhost:9000"
store_access_key_id = "admin"
store_secret_access_key = "password"
store_region = "us-east-1"
store_class = "s3"

[transforms.add_fields]
enabled = true

[[transforms.add_fields.fields]]
key = "db_table"
value.static = "nyc.users"

Transforms are declared at the top level of the connector file, next to [plugin_config], and every [transforms.*] section requires enabled = true. See the transforms documentation for details.

Note: The value in the message field must contain both the namespace and the table name, separated by a dot (.). Example:

• Namespace: nyc
• Table name: users

Source Compatibility

The Iceberg sink expects flat JSON where each top-level key maps directly to a column in the target Iceberg table schema. Sources that wrap row data in an envelope (with metadata fields alongside a nested data object) are not directly compatible: the Arrow JSON reader will map envelope keys to table columns, producing nulls or schema errors.

If your source emits envelope-wrapped JSON, use the unwrap_envelope transform to extract the inner data field before it reaches the sink:

[transforms.unwrap_envelope]
enabled = true
field = "data"

Set field to the envelope key that holds the row data. See the transforms documentation and your source connector's documentation for details on the envelope shape.