Postgre CDC source connector

Support Those Engines

SeaTunnel Zeta
Flink

Key features

• batch
• stream
• exactly-once
• column projection
• parallelism
• support user-defined split

Description

The Postgre CDC connector allows for reading snapshot data and incremental data from Postgre database. This document describes how to set up the Postgre CDC connector to run SQL queries against Postgre databases.

Supported DataSource Info

Using Dependency

Install Jdbc Driver

For Spark/Flink Engine

1. You need to ensure that the jdbc driver jar package has been placed in directory ${SEATUNNEL_HOME}/plugins/.

For SeaTunnel Zeta Engine

1. You need to ensure that the jdbc driver jar package has been placed in directory ${SEATUNNEL_HOME}/lib/.

Please download and put Postgre driver in ${SEATUNNEL_HOME}/lib/ dir. For example: cp postgresql-xxx.jar $SEATNUNNEL_HOME/lib/

Here are the steps to enable CDC (Change Data Capture) in PostgreSQL:

1. Ensure the wal_level is set to logical: Modify the postgresql.conf configuration file by adding “wal_level = logical”, restart the PostgreSQL server for the changes to take effect. Alternatively, you can use SQL commands to modify the configuration directly:

ALTER SYSTEM SET wal_level TO 'logical';
SELECT pg_reload_conf();

1. Change the REPLICA policy of the specified table to FULL

ALTER TABLE your_table_name REPLICA IDENTITY FULL;

Data Type Mapping

Source Options

Task Example

Simple

Support multi-table reading

env {
  # You can set engine configuration here
  execution.parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 5000
  read_limit.bytes_per_second=7000000
  read_limit.rows_per_second=400
}
source {
  Postgres-CDC {
    result_table_name = "customers_Postgre_cdc"
    username = "postgres"
    password = "postgres"
    database-names = ["postgres_cdc"]
    schema-names = ["inventory"]
    table-names = ["postgres_cdc.inventory.postgres_cdc_table_1,postgres_cdc.inventory.postgres_cdc_table_2"]
    base-url = "jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
  }
}
transform {
}
sink {
  jdbc {
    source_table_name = "customers_Postgre_cdc"
    url = "jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
    driver = "org.postgresql.Driver"
    user = "postgres"
    password = "postgres"
    generate_sink_sql = true
    # You need to configure both database and table
    database = postgres_cdc
    chema = "inventory"
    tablePrefix = "sink_"
    primary_keys = ["id"]
  }
}

Support custom primary key for table

source {
  Postgres-CDC {
    result_table_name = "customers_mysql_cdc"
    username = "postgres"
    password = "postgres"
    database-names = ["postgres_cdc"]
    schema-names = ["inventory"]
    table-names = ["postgres_cdc.inventory.full_types_no_primary_key"]
    base-url = "jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
    decoding.plugin.name = "decoderbufs"
    exactly_once = false
    table-names-config = [
      {
        table = "postgres_cdc.inventory.full_types_no_primary_key"
        primaryKeys = ["id"]
      }
    ]
  }
}

Changelog

• Add Postgre CDC Source Connector

next version