> ## 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. # 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──▶ Ingest Engine ──▶ 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 Ingest Engine deployment with the controller, operator, and storage configured * PostgreSQL with `wal_level = logical` and a user with `REPLICATION` privilege — see [CDC connector prerequisites](/ingest/connectors/databases/cdc#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 theme={"theme":{"light":"github-light","dark":"github-dark"}} -- 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 An Ingest Engine 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 Ingest Engine 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 Ingest Engine 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 Ingest Engine which PostgreSQL database to connect to, which tables to capture, and how to handle the initial snapshot. ```yaml theme={"theme":{"light":"github-light","dark":"github-dark"}} 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 publicationName: laminar_cdc_publication 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 theme={"theme":{"light":"github-light","dark":"github-dark"}} 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 theme={"theme":{"light":"github-light","dark":"github-dark"}} 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 theme={"theme":{"light":"github-light","dark":"github-dark"}} 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 Ingest Engine operator will reconcile the resources, create the connection tables via the controller API, and start the pipeline automatically. ```bash theme={"theme":{"light":"github-light","dark":"github-dark"}} 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 theme={"theme":{"light":"github-light","dark":"github-dark"}} 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 Set the API values used by the requests: ```bash theme={"theme":{"light":"github-light","dark":"github-dark"}} export INGESTION_ENGINE_API_URL="https://" export INGESTION_ENGINE_API_TOKEN="" ``` #### Step 1: Create the CDC source Register the PostgreSQL CDC source by posting the connection config to the Ingest Engine API. This creates a connection table that Ingest Engine uses to connect to PostgreSQL and capture changes. ```bash theme={"theme":{"light":"github-light","dark":"github-dark"}} curl -X POST "$INGESTION_ENGINE_API_URL/api/v1/connectors/cdc/tables" \ -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" \ -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", "publicationName": "laminar_cdc_publication" }, "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 theme={"theme":{"light":"github-light","dark":"github-dark"}} { "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. Ingest Engine will auto-create Iceberg tables in this namespace as CDC data arrives. ```bash theme={"theme":{"light":"github-light","dark":"github-dark"}} curl -X POST "$INGESTION_ENGINE_API_URL/api/v1/connectors/iceberg/tables" \ -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" \ -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 theme={"theme":{"light":"github-light","dark":"github-dark"}} { "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 theme={"theme":{"light":"github-light","dark":"github-dark"}} curl -X POST "$INGESTION_ENGINE_API_URL/api/v1/pipelines" \ -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" \ -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 theme={"theme":{"light":"github-light","dark":"github-dark"}} { "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 theme={"theme":{"light":"github-light","dark":"github-dark"}} # REST catalog curl /v1/namespaces/cdc_db/tables ``` **Response:** ```json theme={"theme":{"light":"github-light","dark":"github-dark"}} { "identifiers": [ { "namespace": ["cdc_db"], "name": "orders" }, { "namespace": ["cdc_db"], "name": "customers" } ] } ``` ```bash theme={"theme":{"light":"github-light","dark":"github-dark"}} # AWS Glue aws glue get-tables --database-name cdc_db --query "TableList[].Name" ``` **Response:** ```json theme={"theme":{"light":"github-light","dark":"github-dark"}} ["orders", "customers"] ``` ### Query with pyiceberg ```python theme={"theme":{"light":"github-light","dark":"github-dark"}} 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 theme={"theme":{"light":"github-light","dark":"github-dark"}} 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 theme={"theme":{"light":"github-light","dark":"github-dark"}} -- 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 Ingest Engine 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 theme={"theme":{"light":"github-light","dark":"github-dark"}} "tables": ["orders", "customers", "products", "invoices", "shipments"] ``` To capture all tables in the database, leave the list empty: ```json theme={"theme":{"light":"github-light","dark":"github-dark"}} "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 theme={"theme":{"light":"github-light","dark":"github-dark"}} 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 theme={"theme":{"light":"github-light","dark":"github-dark"}} # Stop and delete pipeline curl -X PATCH "$INGESTION_ENGINE_API_URL/api/v1/pipelines/{pipeline_id}" \ -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" \ -H "Content-Type: application/json" \ -d '{"stop": "graceful"}' curl -X DELETE "$INGESTION_ENGINE_API_URL/api/v1/pipelines/{pipeline_id}" \ -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" # Delete tables curl -X DELETE "$INGESTION_ENGINE_API_URL/api/v1/connection_tables/{source_table_id}" \ -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" curl -X DELETE "$INGESTION_ENGINE_API_URL/api/v1/connection_tables/{sink_table_id}" \ -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" ``` *** ## 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 * Source configuration, snapshot modes, and database prerequisites: [CDC connector reference](/ingest/connectors/databases/cdc). This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.