> For the complete documentation index, see [llms.txt](https://docs.e6data.com/ingestion-engine/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.e6data.com/ingestion-engine/get-started/examples/postgres-cdc-to-iceberg.md). # PostgreSQL CDC to Iceberg Replicate PostgreSQL tables into Apache Iceberg in real time. This covers the full lifecycle: snapshot of existing data, live streaming of WAL changes, and verifying output in the Iceberg catalog. **What you'll build** ``` PostgreSQL (orders, customers) ──CDC──▶ Laminar ──▶ Iceberg catalog ├── cdc_db.orders └── cdc_db.customers ``` * Initial snapshot of all existing rows, then continuous streaming from PostgreSQL's logical replication * One Iceberg table per source table, auto-created on first write * Every change (insert, update, delete) is appended as a new row with an `_op` column indicating the operation type, plus `_ts_ms` and `_timestamp` for event timing * Exactly-once file commits via checkpointed two-phase commit ## Prerequisites * A running Laminar deployment with the controller, operator, and storage configured * PostgreSQL with `wal_level = logical` and a user with `REPLICATION` privilege — see [CDC connector prerequisites](/ingestion-engine/connectors/databases/cdc.md#database-prerequisites) * An Iceberg catalog (REST, AWS Glue, or Hive Metastore) with a target database/namespace * Object storage for Iceberg data files (S3, GCS, or ADLS) ### PostgreSQL configuration Ensure these settings on the PostgreSQL instance: ```sql -- Check WAL level (must be 'logical') SHOW wal_level; -- Grant replication to the CDC user ALTER ROLE cdc_user REPLICATION; -- Create a publication for the tables to capture CREATE PUBLICATION laminar_cdc_publication FOR TABLE orders, customers; ``` *** ## Pipeline Deployment A Laminar CDC pipeline consists of three resources: a **source** (where to read changes from), a **sink** (where to write them), and a **pipeline** (the SQL query connecting them). These can be deployed in two ways: * **Kubernetes Manifest** — declare all three resources as YAML and apply with `kubectl`. The Laminar operator reconciles them and starts the pipeline. This is the recommended approach for production deployments and GitOps workflows. * **REST API** — create each resource individually via HTTP calls to the Laminar controller API. Useful for scripting, automation, and environments where `kubectl` access is not available. Both methods use identical configuration — only the delivery mechanism differs. ### Via Kubernetes Manifest Create a YAML manifest with the following three resources. All three can be placed in a single file separated by `---` for convenience. #### Step 1: CDC Source Define a `LaminarTable` resource for the CDC source. This tells Laminar which PostgreSQL database to connect to, which tables to capture, and how to handle the initial snapshot. ```yaml apiVersion: laminar.stream/v1alpha1 kind: LaminarTable metadata: name: pg-cdc-source namespace: e6data spec: clusterRef: e6data connector: cdc config: name: pg_cdc_src config: transport: mode: jni maxHeap: "512m" database: type: postgres host: port: 5432 user: cdc_user password: database: mydb schema: public tables: [orders, customers] snapshotMode: initial debeziumOverrides: {} channelCapacity: 65536 schema: inferred: true fields: - field_name: _table field_type: type: primitive: String nullable: false - field_name: _schema_epoch field_type: type: primitive: Int64 nullable: false - field_name: _num_rows field_type: type: primitive: Int64 nullable: false - field_name: _offsets field_type: type: primitive: Bytes nullable: true - field_name: _payload field_type: type: primitive: Bytes nullable: false ``` **Key configuration:** | Field | Description | | ------------------- | ------------------------------------------------------------------------------------------------------------- | | `transport.mode` | `jni` runs the CDC engine in-process. Use `grpc` for out-of-process isolation. | | `transport.maxHeap` | JVM heap for the CDC engine. 512m is sufficient for most tables. Increase for wide schemas (200+ columns). | | `snapshotMode` | `initial` — full snapshot then streaming. Other modes: `no_data` (stream-only), `when_needed`, `incremental`. | | `channelCapacity` | Internal buffer between CDC engine and pipeline. Default 65536 is suitable for most workloads. | | `debeziumOverrides` | Pass additional Debezium properties (e.g., `snapshot.max.threads`, `poll.interval.ms`). | No column schema needed for the source tables — column names and types are read from PostgreSQL automatically and carried with each batch of rows. #### Step 2: Iceberg Sink Define a `LaminarTable` resource for the Iceberg sink. This configures the catalog connection, target namespace, multi-table routing, storage credentials, and output format. ```yaml apiVersion: laminar.stream/v1alpha1 kind: LaminarTable metadata: name: pg-cdc-iceberg-sink namespace: e6data spec: clusterRef: e6data connector: iceberg config: name: pg_cdc_iceberg_sink config: catalog: provider: direct format: iceberg # table format type: rest # catalog type: rest, glue, hive, or biglake url: warehouse: namespace: cdc_db tableName: _unused appendOnly: true multiTable: enabled: true tableNameTemplate: "{{table}}" storageOptions: s3.region: us-east-1 s3.access-key-id: s3.secret-access-key: parquet: compression: snappy schema: format: parquet: {} fields: - field_name: _table field_type: type: primitive: String nullable: false - field_name: _schema_epoch field_type: type: primitive: Int64 nullable: false - field_name: _num_rows field_type: type: primitive: Int64 nullable: false - field_name: _payload field_type: type: primitive: Bytes nullable: false ``` **Polaris catalog** — replace the `catalog` and `storageOptions` sections with: ```yaml catalog: provider: direct format: iceberg type: rest url: https:///api/catalog warehouse: auth: # OAuth2 credentials for Polaris catalog API credential: ":" oauth2-server-uri: https:///api/catalog/v1/oauth/tokens scope: PRINCIPAL_ROLE:ALL storageOptions: client.assume-role.arn: "arn:aws:iam:::role/" client.assume-role.session-name: laminar s3.region: ``` Polaris uses OAuth2 for authentication. The `credential` is the client ID and secret separated by a colon. If your data is in a different AWS account, use `client.assume-role.arn` in `storageOptions` for cross-account S3 access. For production deployments, avoid static credentials in the manifest. Instead, use your cloud platform's pod identity mechanism to grant storage access: * **AWS (EKS)**: IAM Roles for Service Accounts (IRSA) — the pod assumes an IAM role with S3 permissions * **GCP (GKE)**: Workload Identity — the pod uses a GCP service account with GCS permissions * **Azure (AKS)**: Workload Identity — the pod uses a managed identity with ADLS permissions When pod identity is configured, the `storageOptions` credentials can be omitted entirely — the pod inherits storage permissions from its service account. The `auth` block for catalog authentication is still required. **Key configuration:** | Field | Description | | ------------------------------ | ------------------------------------------------------------------------------------------------------------- | | `namespace` | Iceberg namespace where tables are created | | `multiTable.enabled` | Routes each source table to its own Iceberg table | | `multiTable.tableNameTemplate` | Template for output table names. `{{table}}` uses the source table name. Use `cdc_{{table}}` to add a prefix. | | `tableName` | Required field, but ignored when `multiTable` is enabled. Use any placeholder value. | | `appendOnly` | `true` for append-only CDC (inserts only, no delete files). `false` for full CDC with equality deletes. | | `storageOptions` | S3/GCS/ADLS credentials. Omit if using IAM roles. | #### Step 3: Pipeline Define a `LaminarPipeline` resource that connects the CDC source to the Iceberg sink using a SQL query. The query forwards the CDC envelope from source to sink, and the sink automatically unwraps and routes each table. ```yaml apiVersion: laminar.stream/v1alpha1 kind: LaminarPipeline metadata: name: pg-cdc-to-iceberg namespace: e6data spec: clusterRef: e6data replicas: 1 config: name: pg_cdc_to_iceberg query: | INSERT INTO pg_cdc_iceberg_sink SELECT _table, _schema_epoch, _num_rows, _payload FROM pg_cdc_src parallelism: 1 checkpointIntervalMicros: 10000000 ``` **Configuration:** | Field | Description | | -------------------------- | ----------------------------------------------------------------------------------------------------------- | | `parallelism` | Number of parallel tasks. Use `1` for CDC pipelines (required for offset consistency). | | `checkpointIntervalMicros` | How often to checkpoint. `10000000` = 10 seconds. Lower values reduce latency but increase commit overhead. | #### Step 4: Apply Manifest Apply the manifest to your Kubernetes cluster. The Laminar operator will reconcile the resources, create the connection tables via the controller API, and start the pipeline automatically. ```bash kubectl apply -f postgres-cdc-to-iceberg.yaml ``` **Output:** ``` namespace/e6data unchanged laminartable.laminar.stream/pg-cdc-source created laminartable.laminar.stream/pg-cdc-iceberg-sink created laminarpipeline.laminar.stream/pg-cdc-to-iceberg created ``` Check reconciliation: ```bash kubectl get laminartables,laminarpipelines -n e6data ``` **Output:** ``` NAME CONNECTOR STATUS laminartable.laminar.stream/pg-cdc-source cdc Ready laminartable.laminar.stream/pg-cdc-iceberg-sink iceberg Ready NAME STATUS PARALLELISM laminarpipeline.laminar.stream/pg-cdc-to-iceberg Running 1 ``` The `config` block inside each resource is identical to the API format — only the Kubernetes wrapper (`apiVersion`, `kind`, `metadata`, `spec.clusterRef`) differs. *** ### Via REST API ### Step 1: Create the CDC source Register the PostgreSQL CDC source by posting the connection config to the Laminar API. This creates a connection table that Laminar uses to connect to PostgreSQL and capture changes. ```bash curl -X POST /api/v1/connectors/cdc/tables \ -H 'Content-Type: application/json' -d '{ "name": "pg_cdc_src", "config": { "transport": { "mode": "jni", "maxHeap": "512m" }, "database": { "type": "postgres", "host": "", "port": 5432, "user": "cdc_user", "password": "", "database": "mydb", "schema": "public", "tables": ["orders", "customers"], "snapshotMode": "initial" }, "debeziumOverrides": {}, "channelCapacity": 65536 }, "schema": { "inferred": true, "fields": [ {"field_name": "_table", "field_type": {"type": {"primitive": "String"}}, "nullable": false}, {"field_name": "_schema_epoch", "field_type": {"type": {"primitive": "Int64"}}, "nullable": false}, {"field_name": "_num_rows", "field_type": {"type": {"primitive": "Int64"}}, "nullable": false}, {"field_name": "_offsets", "field_type": {"type": {"primitive": "Bytes"}}, "nullable": true}, {"field_name": "_payload", "field_type": {"type": {"primitive": "Bytes"}}, "nullable": false} ] } }' ``` **Response:** ```json { "id": "ct_cdc_xxx", "name": "pg_cdc_src", "connector": "cdc", "tableType": "source" } ``` ### Step 2: Create the Iceberg sink Register the Iceberg sink by posting the catalog connection, namespace, multi-table routing, and storage credentials. Laminar will auto-create Iceberg tables in this namespace as CDC data arrives. ```bash curl -X POST /api/v1/connectors/iceberg/tables \ -H 'Content-Type: application/json' -d '{ "name": "pg_cdc_iceberg_sink", "config": { "catalog": { "provider": "direct", "format": "iceberg", "type": "rest", "url": "", "warehouse": "" }, "namespace": "cdc_db", "tableName": "_unused", "appendOnly": true, "multiTable": { "enabled": true, "tableNameTemplate": "{{table}}" }, "storageOptions": { "s3.region": "us-east-1", "s3.access-key-id": "", "s3.secret-access-key": "" }, "parquet": { "compression": "snappy" } }, "schema": { "format": {"parquet": {}}, "fields": [ {"field_name": "_table", "field_type": {"type": {"primitive": "String"}}, "nullable": false}, {"field_name": "_schema_epoch", "field_type": {"type": {"primitive": "Int64"}}, "nullable": false}, {"field_name": "_num_rows", "field_type": {"type": {"primitive": "Int64"}}, "nullable": false}, {"field_name": "_payload", "field_type": {"type": {"primitive": "Bytes"}}, "nullable": false} ] } }' ``` **Response:** ```json { "id": "ct_iceberg_xxx", "name": "pg_cdc_iceberg_sink", "connector": "iceberg", "tableType": "sink" } ``` For Polaris catalog, replace the `catalog` and `storageOptions` sections with the Polaris-specific config shown in the Kubernetes Manifest section above. ### Step 3: Create the pipeline Create the pipeline with a SQL query that connects the CDC source to the Iceberg sink. The pipeline starts automatically once created and begins with a full snapshot of existing data. ```bash curl -X POST /api/v1/pipelines \ -H 'Content-Type: application/json' -d '{ "name": "pg-cdc-to-iceberg", "query": "INSERT INTO pg_cdc_iceberg_sink SELECT _table, _schema_epoch, _num_rows, _payload FROM pg_cdc_src", "parallelism": 1, "checkpointIntervalMicros": 10000000 }' ``` **Response:** ```json { "id": "pl_xxx", "name": "pg-cdc-to-iceberg", "parallelism": 1, "actionText": "Running" } ``` *** ## Watch the snapshot The pipeline begins with a full snapshot of existing data. In worker logs: ``` Snapshot - Final stage Exported 11569 of 11569 records for table 'public.orders' after 00:00:03 Exported 34162 of 34162 records for table 'public.customers' after 00:00:06 Snapshot completed Starting streaming ``` A single-threaded snapshot streams at roughly 200–250k rows/s. Once complete, the pipeline transitions to streaming mode and captures live WAL changes. ## Verify the output ### Check Iceberg tables Tables are auto-created in the configured namespace. Verify via your catalog: ```bash # REST catalog curl /v1/namespaces/cdc_db/tables ``` **Response:** ```json { "identifiers": [ { "namespace": ["cdc_db"], "name": "orders" }, { "namespace": ["cdc_db"], "name": "customers" } ] } ``` ```bash # AWS Glue aws glue get-tables --database-name cdc_db --query "TableList[].Name" ``` **Response:** ```json ["orders", "customers"] ``` ### Query with pyiceberg ```python from pyiceberg.catalog import load_catalog catalog = load_catalog("rest", **{ "type": "rest", "uri": "", "warehouse": "", "s3.endpoint": "", "s3.access-key-id": "", "s3.secret-access-key": "", }) tbl = catalog.load_table("cdc_db.orders") df = tbl.scan().to_pandas() print(df[["id", "customer_id", "total", "_op", "_ts_ms"]].head()) print(f"{len(df)} rows") ``` **Output:** ``` id customer_id total _op _ts_ms 0 1 3 29.99 r 1720000000000 1 2 7 49.99 r 1720000000000 2 3 1 12.50 r 1720000000000 3 4 5 99.00 r 1720000000000 4 5 2 35.00 r 1720000000000 11569 rows ``` ### Query with DuckDB ```sql SELECT _op, count(*) FROM iceberg_scan('s3:////cdc_db/orders') GROUP BY _op; ``` **Output:** ``` ┌──────┬──────────────┐ │ _op │ count_star() │ ├──────┼──────────────┤ │ r │ 11569 │ └──────┴──────────────┘ ``` ## Test streaming Make changes in PostgreSQL and verify they land in Iceberg: ```sql -- Insert INSERT INTO orders (id, customer_id, total) VALUES (1001, 7, 49.99); -- Update UPDATE orders SET total = 59.99 WHERE id = 1001; -- Delete DELETE FROM orders WHERE id = 1001; ``` Each change lands in Iceberg within the checkpoint interval (default 10 seconds). Rows carry: | Column | Description | | ------------ | ----------------------------------------------------------------- | | `_op` | `r` (snapshot read), `c` (insert), `u` (update), `d` (delete) | | `_ts_ms` | Source event timestamp (when the change occurred at the database) | | `_timestamp` | Ingest time (when Laminar processed the event) | ## Snapshot modes | Mode | Behavior | Use case | | ------------------- | ----------------------------------------------- | -------------------------------------------- | | `initial` (default) | Full snapshot of existing rows, then stream | First-time setup | | `no_data` | Skip snapshot, stream from current WAL position | When historical data is not needed | | `when_needed` | Snapshot only if no prior offset exists | Recovery after redeployment | | `incremental` | Chunked reads without table locks | Large tables where locking is not acceptable | Set via `"snapshotMode": "incremental"` in the CDC source config. ## Multi-table setup A single pipeline can capture any number of tables. List the tables to capture in the source config: ```json "tables": ["orders", "customers", "products", "invoices", "shipments"] ``` To capture all tables in the database, leave the list empty: ```json "tables": [] ``` Each table lands in its own Iceberg table in the configured namespace. Table names are derived from the `tableNameTemplate`: | Template | Source table `orders` becomes | | ---------------------- | ----------------------------- | | `{{table}}` | `orders` | | `cdc_{{table}}` | `cdc_orders` | | `{{schema}}_{{table}}` | `public_orders` | ## Cleanup ### Via Kubernetes Manifest ```bash kubectl delete laminarpipeline pg-cdc-to-iceberg -n e6data kubectl delete laminartable pg-cdc-iceberg-sink -n e6data kubectl delete laminartable pg-cdc-source -n e6data ``` ### Via REST API ```bash # Stop and delete pipeline curl -X PATCH /api/v1/pipelines/{pipeline_id} \ -H "Content-Type: application/json" \ -d '{"stop": "graceful"}' curl -X DELETE /api/v1/pipelines/{pipeline_id} # Delete tables curl -X DELETE /api/v1/connection_tables/{source_table_id} curl -X DELETE /api/v1/connection_tables/{sink_table_id} ``` *** ## Production considerations * **Checkpoint interval**: 10 seconds is a good balance. Lower values (1–3s) reduce end-to-end latency but increase Iceberg commit frequency. Higher values (30–60s) reduce commit overhead for high-throughput pipelines. * **JVM heap**: 512m is sufficient for most workloads. Increase to 1–2g for databases with very wide schemas (200+ columns) or large initial snapshots. ## Notes * Tuning, failure modes, and troubleshooting: [CDC connector reference](/ingestion-engine/connectors/databases/cdc.md#troubleshooting). --- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://docs.e6data.com/ingestion-engine/get-started/examples/postgres-cdc-to-iceberg.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.