Sinks

How-to guideApplies to: Unified engine deployments

A sink defines where a pipeline writes its transformed records. This page writes to Iceberg through a REST catalog. The catalog, warehouse storage, and access must be ready as described in Before you start.

Infer the output columns

If you already ran the walkthrough, reuse paid_orders_sink. Otherwise, edit these values and register the sink once.

CREATE TABLE paid_orders_sink WITH (
  connector = 'iceberg',
  type = 'sink',
  'catalog.type' = 'rest',
  'catalog.url' = 'http://lakekeeper:8181/catalog',
  'catalog.warehouse' = 'warehouse-id',
  namespace = 'analytics',
  table = 'paid_orders',
  append_only = 'true',
  infer_schema = 'true'
);

The result contains an accepted table_id. The transformation supplies the sink’s column names and types when infer_schema is enabled. Declare useful aliases for calculated columns in your SELECT. The connector writes to analytics.paid_orders in the selected warehouse. Registering the connector does not publish output rows. Publication happens when a pipeline runs and its sink completes the required checkpoint and commit work.

Declare columns explicitly

Use an explicit schema when you want the sink definition to fix the expected output columns. This alternative points to a separate test table.

CREATE TABLE typed_orders_sink (
  order_id BIGINT,
  customer_id BIGINT,
  amount DOUBLE,
  amount_with_tax DOUBLE
) WITH (
  connector = 'iceberg',
  type = 'sink',
  'catalog.type' = 'rest',
  'catalog.url' = 'http://lakekeeper:8181/catalog',
  'catalog.warehouse' = 'warehouse-id',
  namespace = 'analytics',
  table = 'typed_paid_orders',
  append_only = 'true'
);

Match the transformation’s output to the declared columns. The example has no format option because the Iceberg sink writes files through its own writer.

Connection and write options

Option	Purpose	Default
`catalog.type`	Catalog provider; these examples use `rest`	Required
`catalog.url`	REST catalog endpoint	Required for REST
`catalog.warehouse`	Warehouse identifier understood by the catalog	Unset; catalog dependent
`namespace`	Destination namespace	Set explicitly for these single-table examples
`table`	Destination Iceberg table name	Set explicitly for these single-table examples
`infer_schema`	Take columns from the query when the sink has no declared columns	`false`
`append_only`	Use append output for these insert pipelines	`true`
`parquet.compression`	`uncompressed`, `snappy`, `gzip`, `lz4`, or `zstd`	`uncompressed`
`file_rotation.max_file_size_bytes`	File size threshold for rotation	`536870912` bytes

The file rotation threshold controls file writing. The pipeline’s checkpoint_interval controls how often it requests a checkpoint. Neither setting is a promise that each interval will make new rows visible. These examples append records. They do not define updates or deletes based on a Kafka record’s business key.

Catalog authentication

Add the relevant options to the sink’s WITH clause. The following fragments show the option names; replace their values for your catalog. For a bearer token:

'catalog.auth.token' = '{{ ICEBERG_CATALOG_TOKEN }}'

For OAuth client credentials:

'catalog.auth.credential' = '{{ ICEBERG_CLIENT_CREDENTIAL }}',
'catalog.auth.oauth2_server_uri' = 'https://identity.example.com/oauth/token',
'catalog.auth.scope' = 'catalog'

Use the client credential format and scope required by your catalog. The environment references use Streaming connector expansion. Supply the referenced values to the Streaming processes that resolve the catalog connection. A Unified saved credential is not automatically bound to this sink.

Warehouse storage access

Catalog authentication grants access to catalog operations. The writer also needs access to the files in warehouse storage. Use the runtime’s storage identity or catalog-vended credentials where your catalog supports them. For S3-compatible storage, these option names can be supplied in the sink’s WITH clause when required by your deployment:

's3.endpoint' = 'http://minio:9000',
's3.region' = 'us-east-1',
's3.path-style-access' = 'true'

Use endpoints reachable by the Streaming processes. The endpoint fragment does not supply credentials. Configure the storage identity and permissions for the warehouse as part of the deployment.

Read the destination

paid_orders_sink is the Streaming connection name. Read analytics.paid_orders through an Iceberg catalog connection when checking output. Connecting the sink does not register a read catalog in Unified. Choose the matching catalog recipe for a Unified read connection to the same warehouse.

Clean up

Stop and remove pipeline users before removing the connector definition through the existing table controls. Remove the test Iceberg table separately through its catalog. Deleting a connector definition is not table-data cleanup.

Transformation Engine

Learn Laminar

Workloads

SQL reference

HTTP API

Recipes

Operations

Infer the output columns

Declare columns explicitly

Connection and write options

Catalog authentication

Warehouse storage access

Read the destination

Clean up

​Infer the output columns

​Declare columns explicitly

​Connection and write options

​Catalog authentication

​Warehouse storage access

​Read the destination

​Clean up

Infer the output columns

Declare columns explicitly

Connection and write options

Catalog authentication

Warehouse storage access

Read the destination

Clean up