# PostgreSQL CDC Patterns

> For the complete documentation index, see [llms.txt](https://docs.redpanda.com/llms.txt). Component-specific: [connect-full.txt](https://docs.redpanda.com/connect-full.txt)

---
title: PostgreSQL CDC Patterns
latest-connect-version: 4.104.0
latest-operator-version: v26.2.1
latest-console-tag: v3.9.0
latest-redpanda-tag: v26.2.1
docname: postgres_cdc
page-component-name: connect
page-version: master
page-component-version: master
page-component-title: Connect
page-relative-src-path: postgres_cdc.adoc
page-edit-url: https://github.com/redpanda-data/rp-connect-docs/edit/main/modules/cookbooks/pages/postgres_cdc.adoc
description: Learn how to capture, filter, transform, and route PostgreSQL change data capture (CDC) events with Redpanda Connect.
page-topic-type: cookbook
personas: streaming_developer, data_engineer
learning-objective-1: Apply reusable patterns for capturing PostgreSQL CDC events
learning-objective-2: Adapt integration patterns to route CDC data to Redpanda and S3
learning-objective-3: Identify patterns for filtering and transforming change events
page-git-created-date: "2026-06-17"
page-git-modified-date: "2026-06-17"
---

<!-- Source: https://docs.redpanda.com/connect/cookbooks/postgres_cdc.md -->

The `postgres_cdc` input captures row-level changes from PostgreSQL tables using logical replication and the Write-Ahead Log (WAL). Use these patterns to filter, transform, and route PostgreSQL CDC events to Redpanda, S3, and other destinations.

Use this cookbook to:

-   Apply reusable patterns for capturing PostgreSQL CDC events

-   Adapt integration patterns to route CDC data to Redpanda and S3

-   Identify patterns for filtering and transforming change events


## [](#prerequisites)Prerequisites

Before using these patterns, configure the following.

### [](#redpanda-cli)Redpanda CLI

Install the Redpanda CLI (`rpk`) to run Redpanda Connect. See [rpk installation](https://docs.redpanda.com/connect/get-started/quickstarts/rpk/) for installation instructions.

### [](#postgresql-logical-replication)PostgreSQL logical replication

The source PostgreSQL database must have logical replication enabled. Verify the setting by running:

```sql
SHOW wal_level;
```

The `wal_level` value must be `logical`. For instructions on enabling logical replication, see the [postgres\_cdc prerequisites](https://docs.redpanda.com/connect/components/inputs/postgres_cdc/#_prerequisites). For cloud-managed PostgreSQL, see:

-   [Amazon RDS for PostgreSQL](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/PostgreSQL.Concepts.General.FeatureSupport.LogicalReplication.html)

-   [Azure Database for PostgreSQL](https://learn.microsoft.com/en-us/azure/postgresql/flexible-server/concepts-logical)

-   [Google Cloud SQL for PostgreSQL](https://cloud.google.com/sql/docs/postgres/replication/configure-logical-replication)


### [](#replication-slot)Replication slot

The `postgres_cdc` input uses a PostgreSQL replication slot to track its position in the WAL. The slot is created automatically when the pipeline starts, using the name specified in the `slot_name` field. Each pipeline must use a unique slot name. Replication slots persist in the database, so the pipeline resumes from where it left off after a restart.

### [](#environment-variables)Environment variables

The examples in this cookbook use environment variables for configuration:

```bash
export PG_DSN="postgres://user:password@localhost:5432/mydb" (1)
export PG_SCHEMA="public" (2)
export PG_SLOT_NAME="my_connect_slot" (3)
export REDPANDA_BROKERS=localhost:9092 (4)
export S3_BUCKET=cdc-archive (5)
```

| 1 | The PostgreSQL DSN in postgres://user:password@host:port/dbname format. |
| --- | --- |
| 2 | The schema containing the tables to capture. |
| 3 | A unique name for the replication slot. |
| 4 | The Redpanda broker addresses (for Redpanda output examples). |
| 5 | The S3 bucket name (for S3 output examples). |

## [](#capture-cdc-events)Capture CDC events

The simplest pattern captures all change events from PostgreSQL tables and outputs them with metadata:

```yaml
input:
  postgres_cdc:
    dsn: ${PG_DSN}
    schema: ${PG_SCHEMA}
    tables:
      - orders
    slot_name: ${PG_SLOT_NAME}
    stream_snapshot: true

pipeline:
  processors:
    - mapping: |
        root.operation = meta("operation")
        root.table = meta("table")
        root.lsn = meta("lsn")
        root.data = this
        root.timestamp = now()

output:
  stdout:
    codec: lines
```

For details on the CDC event message structure and available metadata fields, see the [metadata](https://docs.redpanda.com/connect/components/inputs/postgres_cdc/#_metadata) section in the connector reference.

## [](#filter-cdc-events)Filter CDC events

Filter events to process only specific change types using the `operation` metadata field:

```yaml
input:
  postgres_cdc:
    dsn: ${PG_DSN}
    schema: ${PG_SCHEMA}
    tables:
      - orders
    slot_name: ${PG_SLOT_NAME}
    stream_snapshot: false

pipeline:
  processors:
    - mapping: |
        root = if meta("operation") != "insert" && meta("operation") != "update" {
          deleted()
        }
    - mapping: |
        root.operation = meta("operation")
        root.table = meta("table")
        root.data = this
        root.timestamp = now()

output:
  stdout:
    codec: lines
```

This pattern:

-   Filters to only `insert` and `update` operations, dropping `delete` and `read` events

-   Transforms the event to a simplified format with a timestamp


## [](#route-to-redpanda)Route to Redpanda

Stream PostgreSQL changes to Redpanda for real-time processing:

```yaml
input:
  postgres_cdc:
    dsn: ${PG_DSN}
    schema: ${PG_SCHEMA}
    tables:
      - orders
      - customers
    slot_name: ${PG_SLOT_NAME}
    stream_snapshot: true

pipeline:
  processors:
    - mapping: |
        meta topic = meta("table")

output:
  redpanda:
    seed_brokers:
      - ${REDPANDA_BROKERS}
    topic: ${! meta("topic") }
    key: ${! json("id") }
    batching:
      count: 100
      period: 1s
```

This pattern:

-   Uses the table name as the Redpanda topic

-   Batches messages for efficient delivery

-   Sets the message key to the row’s primary key field


## [](#route-to-s3)Route to S3

Archive CDC events to S3 for long-term storage and analytics:

```yaml
input:
  postgres_cdc:
    dsn: ${PG_DSN}
    schema: ${PG_SCHEMA}
    tables:
      - orders
    slot_name: ${PG_SLOT_NAME}
    stream_snapshot: true

pipeline:
  processors:
    - mapping: |
        root.operation = meta("operation")
        root.table = meta("table")
        root.data = this
        root.timestamp = now()

output:
  aws_s3:
    bucket: ${S3_BUCKET}
    path: >-
      cdc/${! meta("table") }/${! timestamp_unix().format_timestamp("2006/01/02/15") }/${! uuid_v4() }.ndjson
    batching:
      count: 1000
      period: 5m
      processors:
        - archive:
            format: lines
```

This pattern:

-   Organizes files by table and time-based partitions (year/month/day/hour)

-   Batches events and archives them as newline-delimited JSON

-   Uses UUID file names to prevent collisions


## [](#route-by-event-type)Route by event type

Route different event types to different destinations:

```yaml
input:
  postgres_cdc:
    dsn: ${PG_DSN}
    schema: ${PG_SCHEMA}
    tables:
      - orders
    slot_name: ${PG_SLOT_NAME}
    stream_snapshot: true

output:
  switch:
    cases:
      - check: meta("operation") == "insert"
        output:
          redpanda:
            seed_brokers:
              - ${REDPANDA_BROKERS}
            topic: postgres.orders.inserts
      - output:
          redpanda:
            seed_brokers:
              - ${REDPANDA_BROKERS}
            topic: postgres.orders.changes
```

This pattern:

-   Routes `insert` events to one Redpanda topic and all other change events to another

-   Supports specialized downstream consumers per operation type


## [](#configure-replication-mode)Configure replication mode

The `postgres_cdc` input supports two replication modes controlled by the `stream_snapshot` field:

-   `stream_snapshot: true`: Captures a full snapshot of existing table data before streaming live changes. Use this when you need a complete initial load.

-   `stream_snapshot: false`: Skips the snapshot and streams only changes from the current WAL position. Use this when you only need new changes going forward.


```yaml
input:
  postgres_cdc:
    dsn: ${PG_DSN}
    schema: ${PG_SCHEMA}
    tables:
      - orders
    slot_name: ${PG_SLOT_NAME}
    stream_snapshot: true
    snapshot_batch_size: 1000 (1)
```

| 1 | Number of rows to read per batch during snapshot processing. |
| --- | --- |

## [](#troubleshoot-common-issues)Troubleshoot common issues

Use these steps to diagnose and fix the most common problems with the `postgres_cdc` input.

### [](#no-events-received)No events received

If no events arrive:

1.  Verify logical replication is enabled:

    ```sql
    SHOW wal_level;
    ```

2.  Grant the required PostgreSQL privileges:

    ```sql
    ALTER ROLE your_user WITH REPLICATION;
    GRANT SELECT ON ALL TABLES IN SCHEMA public TO your_user;
    ```

    > 📝 **NOTE**
    >
    > `ALTER ROLE …​ WITH REPLICATION` requires superuser privileges. Run this statement as a superuser, or have a database administrator run it on your behalf.

3.  Check that `slot_name` is unique and not already in use:

    ```sql
    SELECT slot_name, active FROM pg_replication_slots;
    ```


### [](#snapshot-fails-mid-run)Snapshot fails mid-run

If the snapshot fails, the replication slot has an incomplete record of your database. To maintain data integrity, drop the replication slot and restart the pipeline:

```sql
SELECT pg_drop_replication_slot('my_connect_slot');
```

Replace `my_connect_slot` with your `slot_name` value.

### [](#duplicate-events)Duplicate events

The `postgres_cdc` input provides at-least-once delivery. If the pipeline fails between WAL acknowledgments, events may be re-read on restart. To handle duplicates:

-   Use idempotent processing in downstream systems.

-   Deduplicate using the `lsn` metadata field.

-   Lower `checkpoint_limit` to reduce the window of possible duplicates.


## [](#next-steps)Next steps

-   [PostgreSQL CDC input reference](https://docs.redpanda.com/connect/components/inputs/postgres_cdc/)

-   [Redpanda output](https://docs.redpanda.com/connect/components/outputs/redpanda/)


## Suggested labs

-   [Stream Jira Issues to Redpanda for Real-Time Metrics](https://docs.redpanda.com/labs/docker-compose/jira-metrics-pipeline/)

[Search all labs](https://docs.redpanda.com/labs)