Definition
- CDC (Change Data Capture) – Streaming row-level changes from a database's transaction log into Flink as a changelog.
- Direct CDC source – Connectors like
mysql-cdcthat embed Debezium, snapshot the table, and tail the binlog. - Indirect CDC – Reading Debezium or Canal change events from Kafka using
debezium-jsonorcanal-jsonformat. - RowKind semantics – Each change is INSERT, UPDATE_BEFORE + UPDATE_AFTER, or DELETE, enabling correct retractions.
- Primary key – Required on the CDC table so Flink knows which row an update or delete refers to.
The National Bank of India keeps account balances in its main database. Every time Rajesh's balance changes, the database writes that change into its internal journal. CDC is like placing a live reader next to that journal: the moment a balance changes or an account is opened or closed, the reader streams that exact change into Flink within seconds, without ever re-scanning the whole database.
Understanding the Concepts
What is CDC in Flink SQL?
CDC (Change Data Capture) means consuming the stream of row-level changes from a database and turning them into a changelog-backed dynamic table in Flink SQL. You can then filter, join, aggregate, and write the results to a sink for real-time ETL.
-- Direct CDC source: embed Debezium, tail MySQL binlog
CREATE TABLE accounts (
account_id STRING,
holder_name STRING,
balance DECIMAL(15, 2),
status STRING,
updated_at TIMESTAMP(3),
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'
);
What is the difference between direct and indirect CDC?
Direct CDC embeds Debezium inside the Flink source. Indirect CDC reads change events already published to Kafka by a standalone Debezium or Canal connector.
-- Direct: Flink connects to MySQL binlog
CREATE TABLE direct_accounts (
account_id STRING, balance DECIMAL(15, 2),
PRIMARY KEY (account_id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = 'mysql.bank.internal',
'database-name' = 'nbi',
'table-name' = 'accounts'
);
-- Indirect: Debezium publishes to Kafka, Flink reads from Kafka
CREATE TABLE indirect_accounts (
account_id STRING, balance DECIMAL(15, 2),
PRIMARY KEY (account_id) NOT ENFORCED
) WITH (
'connector' = 'kafka',
'topic' = 'dbserver1.nbi.accounts',
'properties.bootstrap.servers' = 'kafka.bank.internal:9092',
'format' = 'debezium-json'
);
Direct CDC: MySQL binlog → Flink (embedded Debezium) Indirect CDC: MySQL binlog → Debezium → Kafka → Flink (debezium-json format)
How does Flink interpret INSERT/UPDATE/DELETE events?
Flink models every change with a RowKind. An insert produces an INSERT row. A delete produces a DELETE row. An update produces an UPDATE_BEFORE row carrying the old values and an UPDATE_AFTER row carrying the new values.
Database operations → Flink RowKind:
INSERT INTO accounts (ACCT-100, 'Rajesh', 100000)
→ +I (ACCT-100, Rajesh, 100000)
UPDATE accounts SET balance = 95000 WHERE account_id = 'ACCT-100'
→ -U (ACCT-100, Rajesh, 100000) ← old values
→ +U (ACCT-100, Rajesh, 95000) ← new values
DELETE FROM accounts WHERE account_id = 'ACCT-100'
→ -D (ACCT-100, Rajesh, 95000)
package com.bank.flink;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
public class CdcInterpretationDemo {
public static void main(String[] args) {
TableEnvironment tEnv = TableEnvironment.create(
EnvironmentSettings.inStreamingMode());
// Register CDC source table
// ...
// Aggregation correctly handles UPDATE_BEFORE/AFTER:
// When Rajesh's balance changes from 100000 to 95000,
// the branch total retracts 100000 and adds 95000
tEnv.executeSql(
"SELECT branch_id, SUM(balance) AS total_balance " +
"FROM accounts_cdc " +
"GROUP BY branch_id"
);
}
}
Why is the UPDATE_BEFORE/AFTER pair essential?
The pair ensures that downstream aggregations stay correct. Without the before image, an aggregation summing balances would not know which old value to subtract.
Branch total calculation:
Initial: ACCT-100 balance = 100,000
ACCT-200 balance = 200,000
Branch total = 300,000
Update: ACCT-100 balance changes to 95,000
-U (ACCT-100, 100,000) → branch total = 300,000 - 100,000 = 200,000
+U (ACCT-100, 95,000) → branch total = 200,000 + 95,000 = 295,000
Correct result: 295,000 = 95,000 + 200,000
How does the debezium-json format map operations?
The debezium-json format maps Debezium's op field to Flink RowKinds.
Debezium op field → Flink RowKind: 'c' (create) → +I (INSERT) 'r' (read/snapshot) → +I (INSERT) 'u' (update) → -U (UPDATE_BEFORE) + +U (UPDATE_AFTER) 'd' (delete) → -D (DELETE)
-- Indirect CDC with debezium-json decoding
CREATE TABLE debezium_accounts (
account_id STRING,
balance DECIMAL(15, 2),
status STRING,
PRIMARY KEY (account_id) NOT ENFORCED
) WITH (
'connector' = 'kafka',
'topic' = 'dbserver1.nbi.accounts',
'properties.bootstrap.servers' = 'kafka.bank.internal:9092',
'format' = 'debezium-json',
'debezium-json.schema-include' = 'false'
);
What about the snapshot phase?
The direct CDC connector first takes a consistent snapshot of the table, emitting existing rows as inserts, then tails the binlog for subsequent changes. Modern connectors use incremental snapshots that chunk the table and avoid long locks.
CDC two-phase operation:
Phase 1: Snapshot
┌──────────────────────────────────┐
│ Read all existing rows │
│ Emit each as +I (INSERT) │
│ Incremental: chunk by PK ranges │
│ No long table lock │
└──────────────────────────────────┘
│
▼
Phase 2: Streaming
┌──────────────────────────────────┐
│ Tail binlog from snapshot point │
│ Emit INSERT, UPDATE, DELETE │
│ Runs continuously │
└──────────────────────────────────┘
How does exactly-once work with CDC?
Exactly-once depends on Flink checkpointing the binlog offset together with operator state. On recovery, the job resumes from the exact checkpointed position.
package com.bank.flink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.CheckpointingMode;
public class CdcExactlyOnceDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(30000, CheckpointingMode.EXACTLY_ONCE);
// Flink checkpoints the binlog offset (e.g., mysql-bin.000003:12345)
// together with operator state.
// On crash recovery:
// 1. Restore operator state from checkpoint
// 2. Resume reading binlog from checkpointed offset
// 3. No change is missed or duplicated
}
}
What should a senior engineer know?
- The CDC table must declare a primary key because the changelog is keyed by it.
- Downstream sinks must be upsert-capable because CDC is a changelog, not an append log.
- Retain the binlog long enough and grant replication privileges, or the connector cannot resume after downtime.
- Schema evolution in the source database must be handled; newer connectors support schema-change propagation.
- The initial snapshot can be heavy, so use incremental chunked snapshots for large tables.
-- Complete CDC pipeline: source → transform → upsert sink
CREATE TABLE account_mirror (
account_id STRING, balance DECIMAL(15, 2), status STRING,
PRIMARY KEY (account_id) NOT ENFORCED
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:postgresql://analytics.bank.internal:5432/mirror',
'table-name' = 'accounts'
);
INSERT INTO account_mirror
SELECT account_id, balance, status
FROM accounts_cdc
WHERE status != 'CLOSED';
-- Inserts, updates, and deletes from MySQL replicate to PostgreSQL in real time
Visual Overview
CDC Data Flow
┌──────────────┐
│ MySQL DB │
│ (accounts) │
└──────┬───────┘
│ binlog
▼
┌──────────────────┐
│ CDC Connector │
│ (Debezium) │
│ Snapshot → Stream│
└──────┬───────────┘
│ changelog events
▼
┌──────────────────────────────┐
│ Flink SQL Dynamic Table │
│ +I (INSERT) │
│ -U (UPDATE_BEFORE) │
│ +U (UPDATE_AFTER) │
│ -D (DELETE) │
└──────┬───────────────────────┘
│ transform / aggregate
▼
┌──────────────────┐
│ Upsert Sink │
│ (JDBC / Kafka) │
│ keyed by PK │
└──────────────────┘
UPDATE_BEFORE / UPDATE_AFTER Pair
Database: UPDATE accounts SET balance = 95000 WHERE id = 'ACCT-100'
Flink receives:
┌────────────────────────────────────┐
│ -U (ACCT-100, balance = 100,000) │ ← old values (retract)
│ +U (ACCT-100, balance = 95,000) │ ← new values (insert)
└────────────────────────────────────┘
Downstream SUM(balance):
Before: 300,000
Retract: 300,000 - 100,000 = 200,000
Insert: 200,000 + 95,000 = 295,000 ← correct
Key Takeaway
Reading CDC data in Flink SQL means consuming a database's stream of row-level inserts, updates, and deletes and turning it into a changelog-backed dynamic table. You can use a direct CDC source like mysql-cdc that embeds Debezium, snapshots the table, and tails the binlog, or you can read Debezium change events from Kafka using the debezium-json format. Either way, Flink keeps the table continuously in sync with the source.
The key insight is that CDC is a changelog, not an append log. An update arrives as an UPDATE_BEFORE with old values and an UPDATE_AFTER with new values, so downstream aggregations retract the old contribution and add the new one correctly. The CDC table must declare a primary key, and the sink must be upsert-capable.
At the National Bank of India we replaced a stale nightly batch extract with CDC so fraud and analytics systems saw account changes within seconds. The connector snapshotted the accounts table once, then streamed every change live from the binlog. When Rajesh's balance changed, the branch total retracted the old value and added the new one, staying correct. Exactly-once came from checkpointing the binlog offset with operator state.
Remember one sentence: CDC turns a database's change journal into a live changelog table in Flink, where updates arrive as before-and-after pairs so aggregations retract correctly, and the sink must be upsert-capable because it is a changelog, not an append log.
---