Definition
- Flink CDC – A family of source connectors built on Debezium that capture row-level changes from operational databases.
- Snapshot phase – Reads all existing rows for a consistent baseline, emitting each as an INSERT.
- Streaming phase – Tails the database's transaction log (binlog for MySQL, WAL for PostgreSQL) for live changes.
- Incremental snapshot – Chunks the table by primary key ranges, snapshots in parallel without long locks.
- Changelog semantics – Changes arrive as INSERT, UPDATE_BEFORE + UPDATE_AFTER, and DELETE with proper RowKind.
The National Bank of India keeps its authoritative records in an operational database. Older systems copied that data every night with a big batch job, so analytics were always a day stale. Flink CDC is like placing a live reader next to the database's change journal: the moment Rajesh's balance changes or Priya opens an account, the reader streams that exact change into Flink within seconds, without ever re-scanning the whole database.
Understanding the Concepts
What is Flink CDC?
Flink CDC is a family of source connectors that capture row-level changes from databases like MySQL, PostgreSQL, and MongoDB and stream them into Flink as a changelog. It replaces stale nightly batch extracts with real-time synchronization.
-- Flink CDC source for MySQL accounts table
CREATE TABLE accounts_cdc (
account_id STRING,
holder_name STRING,
balance DECIMAL(15, 2),
branch_id STRING,
status STRING,
PRIMARY KEY (account_id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = 'mysql.bank.internal',
'port' = '3306',
'username' = 'cdc_reader',
'password' = '${CDC_PASSWORD}',
'database-name' = 'nbi',
'table-name' = 'accounts'
);
How does the two-phase operation work?
Each CDC connector operates in two phases: snapshot first, then streaming.
package com.bank.flink;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
public class FlinkCdcTwoPhaseDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(30000);
MySqlSource<String> source = MySqlSource.<String>builder()
.hostname("mysql.bank.internal")
.port(3306)
.databaseList("nbi")
.tableList("nbi.accounts")
.username("cdc_reader")
.password(System.getenv("CDC_PASSWORD"))
.deserializer(new JsonDebeziumDeserializationSchema())
.build();
// Phase 1: Snapshot existing rows as INSERTs
// Phase 2: Tail binlog for live INSERT/UPDATE/DELETE
env.fromSource(source, WatermarkStrategy.noWatermarks(), "MySQL-CDC")
.print();
env.execute("NBI-CDC-Pipeline");
}
}
Phase 1 (Snapshot): Read all existing accounts → emit as +I (INSERT) rows ACCT-100: +I (ACCT-100, Rajesh, 100000, BR-001, ACTIVE) ACCT-200: +I (ACCT-200, Priya, 250000, BR-002, ACTIVE) ... Phase 2 (Streaming): Tail binlog from snapshot end point Balance update: -U (ACCT-100, 100000) → +U (ACCT-100, 95000) New account: +I (ACCT-300, Pankaj, 500000, BR-001, ACTIVE) Account closed: -D (ACCT-200, Priya, 250000, BR-002, CLOSED)
What is incremental snapshot?
Modern CDC connectors use the incremental snapshot algorithm to avoid long locks on large tables. It splits the table into chunks by primary key ranges and snapshots them in parallel.
Incremental snapshot for accounts table (1M rows): Chunk 1: account_id ACCT-000001 to ACCT-100000 → parallel read Chunk 2: account_id ACCT-100001 to ACCT-200000 → parallel read ... Chunk 10: account_id ACCT-900001 to ACCT-999999 → parallel read Benefits: - No global table lock - Checkpointing during snapshot (can restart mid-snapshot) - Parallelism speeds up large tables - Seamless merge with ongoing binlog reading
How do you build a real-time ETL pipeline with CDC?
CDC enables real-time ETL: capture changes, transform, and write to a sink.
-- Source: MySQL CDC
CREATE TABLE accounts_cdc (
account_id STRING, holder_name STRING,
balance DECIMAL(15, 2), branch_id STRING, status STRING,
PRIMARY KEY (account_id) NOT ENFORCED
) WITH ('connector' = 'mysql-cdc', 'hostname' = 'mysql.bank.internal',
'database-name' = 'nbi', 'table-name' = 'accounts');
-- Sink: Elasticsearch for search
CREATE TABLE accounts_search (
account_id STRING, holder_name STRING,
balance DECIMAL(15, 2), branch_id STRING,
PRIMARY KEY (account_id) NOT ENFORCED
) WITH ('connector' = 'elasticsearch-7',
'hosts' = 'http://es.bank.internal:9200',
'index' = 'accounts');
-- Real-time ETL: filter active accounts, sync to Elasticsearch
INSERT INTO accounts_search
SELECT account_id, holder_name, balance, branch_id
FROM accounts_cdc
WHERE status = 'ACTIVE';
How does exactly-once work?
Exactly-once depends on Flink checkpointing the binlog offset together with operator state. On recovery, the job resumes from the exact checkpointed position.
Exactly-once guarantee:
Checkpoint N:
Binlog position: mysql-bin.000003:12345
Operator state: account ACCT-100 balance = 95000
Crash and recovery:
Restore operator state from Checkpoint N
Resume binlog from mysql-bin.000003:12345
No change missed, no change duplicated
What databases does Flink CDC support?
Supported databases: MySQL → mysql-cdc (reads binlog) PostgreSQL → postgres-cdc (reads WAL via logical replication slot) MongoDB → mongodb-cdc (reads change streams) Oracle → oracle-cdc (reads redo log) SQL Server → sqlserver-cdc (reads change tables) TiDB → tidb-cdc (reads TiKV change feed)
What are common mistakes?
-- WRONG: append-only sink for CDC data
CREATE TABLE bad_sink (
account_id STRING, balance DECIMAL(15, 2)
) WITH ('connector' = 'filesystem', 'path' = '/data/output', 'format' = 'csv');
INSERT INTO bad_sink SELECT account_id, balance FROM accounts_cdc;
-- ERROR: CDC produces retractions, but filesystem is append-only
-- CORRECT: upsert-capable sink
CREATE TABLE good_sink (
account_id STRING, balance DECIMAL(15, 2),
PRIMARY KEY (account_id) NOT ENFORCED
) WITH ('connector' = 'upsert-kafka', 'topic' = 'nbi-accounts-mirror',
'key.format' = 'json', 'value.format' = 'json');
What should a senior engineer know?
- Retain the transaction log long enough and grant replication privileges, or the connector cannot resume.
- Schema evolution in the source must be handled; newer pipeline connectors propagate schema changes automatically.
- CDC is a changelog, not an append log, so sinks must apply changes by primary key.
- Monitor the snapshot progress for large tables; incremental snapshots report chunk completion.
- The initial snapshot can take hours for very large tables; plan accordingly during first deployment.
// Senior tip: monitor CDC lag
// Metric: currentBinlogOffset vs latestBinlogOffset
// If the gap grows, the pipeline is falling behind
// Scale up parallelism or optimize downstream operators
Visual Overview
Flink CDC Two-Phase Operation
┌──────────────┐
│ MySQL DB │
│ (accounts) │
└──────┬───────┘
│
Phase 1: Snapshot
┌──────────────────────────────┐
│ Chunk 1: ACCT-001..100000 │──▶ +I rows
│ Chunk 2: ACCT-100001..200K │──▶ +I rows
│ ... │──▶ +I rows
│ (parallel, no locks) │
└──────────────────────────────┘
│
Phase 2: Streaming
┌──────────────────────────────┐
│ Tail binlog from snapshot │
│ +I (new accounts) │
│ -U/+U (balance changes) │
│ -D (account closures) │
│ (runs continuously) │
└──────────┬───────────────────┘
│
▼
┌──────────────────────────────┐
│ Flink SQL / DataStream │
│ Transform, Join, Aggregate │
└──────────┬───────────────────┘
│
▼
┌──────────────────────────────┐
│ Upsert Sink (Kafka/JDBC/ES) │
│ keyed by account_id │
└──────────────────────────────┘
Key Takeaway
Flink CDC is a family of source connectors built on Debezium that capture row-level changes from databases like MySQL, PostgreSQL, and MongoDB and stream them into Flink as a changelog. Each connector works in two phases: it snapshots the table for a consistent baseline, then tails the transaction log for live changes. Modern connectors use incremental snapshots that chunk the table in parallel without locks.
Changes arrive with proper RowKind semantics: inserts, update-before/after pairs, and deletes. This enables real-time ETL where you transform and write to upsert-capable sinks. Exactly-once comes from checkpointing the transaction log offset with operator state, so after a crash the job resumes from the exact position.
At the National Bank of India we replaced a stale nightly batch extract with Flink CDC so fraud and analytics systems saw account changes within seconds. The connector snapshotted accounts once for a baseline and then streamed every change live from the binlog. We wrote through upsert sinks keyed by account_id so inserts, updates, and deletes applied idempotently. After a crash, the job resumed from the checkpointed binlog offset with no loss or duplication.
Remember one sentence: Flink CDC snapshots a database table and then tails its transaction log to stream every insert, update, and delete into Flink as a real-time changelog, replacing stale batch extracts with continuous synchronization.
---