> ## Documentation Index
> Fetch the complete documentation index at: https://docs.e6data.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Sinks

> 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 desc

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](https://docs.e6data.com/transformation/streaming/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.

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
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.

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
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:

```text theme={"theme":{"light":"github-light","dark":"github-dark"}}
'catalog.auth.token' = '{{ ICEBERG_CATALOG_TOKEN }}'
```

For OAuth client credentials:

```text theme={"theme":{"light":"github-light","dark":"github-dark"}}
'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:

```text theme={"theme":{"light":"github-light","dark":"github-dark"}}
'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](https://docs.e6data.com/transformation/recipes/overview.md#catalog-connections) 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.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.