Apache Flink

<!-- Maven configuration -->
<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <version>42.5.0</version>
</dependency>

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
// Define data stream
DataStream<User> userStream = env.addSource(...);
 
// Configure JDBC connection options
JdbcConnectionOptions jdbcOpts = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
    .withUrl("jdbc:postgresql://tacnode-host:5432/dbname")
    .withDriverName("org.postgresql.Driver")
    .withUsername("your-username")
    .withPassword("your-password")
    .build();
 
// Create JDBC Sink
userStream.addSink(JdbcSink.sink(
    "INSERT INTO users (id, name, age) VALUES (?, ?, ?)",
    (ps, user) -> {
        ps.setInt(1, user.getId());
        ps.setString(2, user.getName());
        ps.setInt(3, user.getAge());
    },
    jdbcOpts));
 
env.execute("Tacnode Sink Job");

JdbcExecutionOptions execOpts = new JdbcExecutionOptions.Builder()
    .withBatchSize(1000)              // Records per batch
    .withBatchIntervalMs(200)         // Batch interval (ms)
    .withMaxRetries(3)                // Max retry attempts
    .build();
 
userStream.addSink(JdbcSink.sink(
    "INSERT INTO users (id, name, age) VALUES (?, ?, ?)",
    (ps, user) -> {
        ps.setInt(1, user.getId());
        ps.setString(2, user.getName());
        ps.setInt(3, user.getAge());
    },
    execOpts,
    jdbcOpts));

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
 
// Register Tacnode Catalog
tableEnv.executeSql(
    "CREATE CATALOG Tacnode WITH (" +
    "  'type'='jdbc'," +
    "  'default-database'='dbname'," +
    "  'username'='your-username'," +
    "  'password'='your-password'," +
    "  'base-url'='jdbc:postgresql://tacnode-host:5432'" +
    ")");
 
// Set active catalog
tableEnv.useCatalog("Tacnode");

// Register Flink source table
tableEnv.executeSql(
    "CREATE TABLE source_table (" +
    "  id INT," +
    "  name STRING," +
    "  age INT" +
    ") WITH (" +
    "  'connector' = 'kafka'," +
    "  'topic' = 'user_topic'," +
    "  'properties.bootstrap.servers' = 'kafka:9092'," +
    "  'format' = 'json'" +
    ")");
 
// Perform insertion
tableEnv.executeSql(
    "INSERT INTO users " +
    "SELECT id, name, age FROM source_table");

CREATE TEMPORARY TABLE users_sink (
 id INTEGER,
 name STRING,
 age INTEGER
) WITH (
 'connector' = 'jdbc',
 'url' = 'jdbc:postgresql://tacnode-host:5432/dbname',
 'table-name' = 'users',
 'username' = 'your-username',
 'password' = 'your-password',
 'driver' = 'org.postgresql.Driver',
 'sink.buffer-flush.max-rows' = '1000',
 'sink.buffer-flush.interval' = '1s',
 'sink.max-retries' = '3'
);
 
INSERT INTO users_sink
SELECT * FROM source_table;

SELECT * FROM pg_create_logical_replication_slot('flink_cdc_slot', 'pgoutput', false, true);

SELECT pg_drop_replication_slot('flink_cdc_slot');

CREATE TEMPORARY TABLE source_cdc (
        ins_time timestamp(6) NULL,
        store_id string NULL,
        store_code string NOT NULL,
        store_name string NULL,
        time_section_no bigint NULL,
        stat_date string NOT NULL,
        PRIMARY KEY (store_code,stat_date) NOT ENFORCED
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = 'host-url',
  'port' = '5432',
  'username' = 'xxxx',
  'password' = 'xxxx',
  'database-name' = 'dbname',
  'schema-name' = 'schemaname',
  'table-name' = 'tablename',
  -- It is recommended to manually create the slot in the database and handle its lifecycle. Without manual creation, Flink CDC automatically creates the slot at startup; automatically created slots lack the NOEXPORT_SNAPSHOT property and may use extra storage.
  'slot.name' = 'flink_cdc_slot',
  -- Decoding plugin required; only pgoutput is supported.
  'decoding.plugin.name' = 'pgoutput',
  -- Currently only supports 'never', which syncs only real-time changes from the slot’s creation, without performing a full snapshot. Full data reading is not supported in the current engine.
  'debezium.snapshot.mode' = 'never',
  -- When using pgoutput decoding, a specific publication can be assigned (optional). If omitted, dbz_publication is automatically created, subscribing to all table updates.
  'debezium.pulication.name' = 'pub-name'
);
 
CREATE TEMPORARY TABLE sink_kafka(
        store_id string NULL,
        cnt bigint NOT NULL,
        PRIMARY KEY (store_id) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'topic' = 'topic-name',
  'properties.bootstrap.servers' = 'servers',
  'key.format' = 'json',
  'key.fields' = 'store_id',
  'value.format' =  'debezium-json',
  'value.debezium-json.schema-include' = 'false'
);
 
INSERT INTO sink_kafka
SELECT store_id, count(*) as cnt
FROM source_cdc GROUP BY store_id;

# log4j.properties
log4j.logger.org.apache.flink.table.runtime=INFO
log4j.logger.org.postgresql=INFO

CREATE TABLE Tacnode_sink (
  id INT,
  name VARCHAR(100),        -- Set length explicitly
  create_time TIMESTAMP(3)
) WITH (...);

INSERT INTO Tacnode_sink
SELECT id, name, CAST(create_time AS TIMESTAMP(3)) FROM source_table;