# 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](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](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

<div class="contain-inline-size rounded-2xl relative bg-token-sidebar-surface-primary" id="bkmrk-create-table-kafka_u"><div class="sticky top-9"><div class="absolute end-0 bottom-0 flex h-9 items-center pe-2"><div class="bg-token-bg-elevated-secondary text-token-text-secondary flex items-center gap-4 rounded-sm px-2 font-sans text-xs">  
</div></div></div><div class="overflow-y-auto p-4" dir="ltr">`<span class="hljs-keyword">CREATE</span> <span class="hljs-keyword">TABLE</span> kafka_users (  id <span class="hljs-type">BIGINT</span>,  name STRING,  email STRING,  created_at <span class="hljs-type">TIMESTAMP</span>(<span class="hljs-number">3</span>),  WATERMARK <span class="hljs-keyword">FOR</span> created_at <span class="hljs-keyword">AS</span> created_at <span class="hljs-operator">-</span> <span class="hljs-type">INTERVAL</span> <span class="hljs-string">'5'</span> <span class="hljs-keyword">SECOND</span>) <span class="hljs-keyword">WITH</span> (  <span class="hljs-string">'connector'</span><span class="hljs-operator">=</span><span class="hljs-string">'kafka'</span>,  <span class="hljs-string">'topic'</span><span class="hljs-operator">=</span><span class="hljs-string">'pg.public.users'</span>,  <span class="hljs-string">'properties.bootstrap.servers'</span><span class="hljs-operator">=</span><span class="hljs-string">'192.168.1.226:9092'</span>,  <span class="hljs-string">'properties.group.id'</span><span class="hljs-operator">=</span><span class="hljs-string">'flink-users-group'</span>,  <span class="hljs-string">'scan.startup.mode'</span><span class="hljs-operator">=</span><span class="hljs-string">'earliest-offset'</span>,  <span class="hljs-string">'format'</span><span class="hljs-operator">=</span><span class="hljs-string">'json'</span>,  <span class="hljs-string">'json.fail-on-missing-field'</span><span class="hljs-operator">=</span><span class="hljs-string">'false'</span>,  <span class="hljs-string">'json.ignore-parse-errors'</span><span class="hljs-operator">=</span><span class="hljs-string">'true'</span>);`</div></div>---

## ⭐ ClickHouse Sink via JDBC

<div class="contain-inline-size rounded-2xl relative bg-token-sidebar-surface-primary" id="bkmrk-create-table-clickho"><div class="sticky top-9"><div class="absolute end-0 bottom-0 flex h-9 items-center pe-2"><div class="bg-token-bg-elevated-secondary text-token-text-secondary flex items-center gap-4 rounded-sm px-2 font-sans text-xs">  
</div></div></div><div class="overflow-y-auto p-4" dir="ltr">`<span class="hljs-keyword">CREATE</span> <span class="hljs-keyword">TABLE</span> clickhouse_users (  id <span class="hljs-type">BIGINT</span>,  name STRING,  email STRING,  created_at <span class="hljs-type">TIMESTAMP</span>(<span class="hljs-number">3</span>)) <span class="hljs-keyword">WITH</span> (    <span class="hljs-string">'connector'</span> <span class="hljs-operator">=</span> <span class="hljs-string">'jdbc'</span>,    <span class="hljs-string">'url'</span> <span class="hljs-operator">=</span> <span class="hljs-string">'jdbc:clickhouse://192.168.1.228:8123/userdb'</span>,    <span class="hljs-string">'table-name'</span> <span class="hljs-operator">=</span> <span class="hljs-string">'users'</span>,    <span class="hljs-string">'username'</span> <span class="hljs-operator">=</span> <span class="hljs-string">'default'</span>,    <span class="hljs-string">'password'</span> <span class="hljs-operator">=</span> <span class="hljs-string">''</span>);`</div></div>---

## ⭐ Streaming insert

<div class="contain-inline-size rounded-2xl relative bg-token-sidebar-surface-primary" id="bkmrk-insert-into-clickhou"><div class="sticky top-9"><div class="absolute end-0 bottom-0 flex h-9 items-center pe-2"><div class="bg-token-bg-elevated-secondary text-token-text-secondary flex items-center gap-4 rounded-sm px-2 font-sans text-xs">  
</div></div></div><div class="overflow-y-auto p-4" dir="ltr">`<span class="hljs-keyword">INSERT</span> <span class="hljs-keyword">INTO</span> clickhouse_users<span class="hljs-keyword">SELECT</span> id, name, email, created_at<span class="hljs-keyword">FROM</span> kafka_users;`</div></div>---

# Jar create

mvn clean package -DskipTests

mvn clean install