CDC streaming pipeline from PostgreSQL → Kafka → Flink → ClickHouse.

CDC streaming pipeline from PostgreSQL → Kafka → Flink → ClickHouse.

sudo wget https://repo1.maven.org/maven2/com/clickhouse/flink/flink-connector-clickhouse-2.0.0/0.1.1/flink-connector-clickhouse-2.0.0-0.1.1-all.jar

wget https://repo1.maven.org/maven2/org/apache/flink/flink-connector-kafka/3.1.0-1.18/flink-connector-kafka-3.1.0-1.18.jar

sudo /opt/flink/bin/flink run flink-clickhouse-job-1.0-SNAPSHOT.jar

/opt/flink/bin/sql-client.sh

ADD JAR 'file:///opt/flink/lib/flink-connector-clickhouse-2.0.0-0.1.1-all.jar';

ADD JAR 'file:///opt/flink/lib/flink-connector-kafka-3.1.0-1.18.jar';
ADD JAR 'file:///opt/flink/lib/clickhouse-jdbc-0.6.0.jar';

ADD JAR 'file:///opt/flink/lib/flink-connector-jdbc-3.1.2-1.18.jar';

⭐ Kafka Source


CREATE TABLE kafka_users ( id BIGINT, name STRING, email STRING, created_at TIMESTAMP(3), WATERMARK FOR created_at AS created_at - INTERVAL '5' SECOND ) WITH ( 'connector'='kafka', 'topic'='pg.public.users', 'properties.bootstrap.servers'='192.168.1.226:9092', 'properties.group.id'='flink-users-group', 'scan.startup.mode'='earliest-offset', 'format'='json', 'json.fail-on-missing-field'='false', 'json.ignore-parse-errors'='true' );

⭐ ClickHouse Sink via JDBC


CREATE TABLE clickhouse_users ( id BIGINT, name STRING, email STRING, created_at TIMESTAMP(3) ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:clickhouse://192.168.1.228:8123/userdb', 'table-name' = 'users', 'username' = 'default', 'password' = '' );

⭐ Streaming insert


INSERT INTO clickhouse_users SELECT id, name, email, created_at FROM kafka_users;

Jar create

mvn clean package -DskipTests

mvn clean install