“

Series: S8 Technology Deep Dives · Article 6 | Level: Advanced | Reading Time: 20 min

Flink CDC 10 Best Practices: The Real-Time Bridge from Database to Lakehouse

#TL;DR

Flink CDC is the core real-time data integration engine in coomia-dip, enabling sub-second synchronization from source databases (MySQL/PostgreSQL) to the Lakehouse (Iceberg) and analytical engines (Doris)
This article presents 10 production-grade best practices covering connector selection, snapshot strategies, schema evolution, exactly-once semantics, monitoring, and Flink CDC 3.0 new features
Each practice includes complete code examples, performance benchmark data, and common pitfall analysis

#1. Practice 1: Choose the Right CDC Mode

#1.1 Three CDC Modes Compared

Mode	Mechanism	Latency	Source DB Impact	Use Case
Query-based	Periodic full-table scans	Minutes	High (table locks)	Legacy systems, small tables
Log-based	Parse Binlog/WAL	Sub-second	Minimal	Production (preferred)
Trigger-based	Database triggers	Milliseconds	Medium (write amplification)	Special cases

coomia-dip's choice: Log-based CDC (via Debezium), because:

Minimal performance impact on source databases (reads Binlog only)
Captures all change types (INSERT/UPDATE/DELETE)
Supports seamless full-snapshot-to-incremental transitions

#1.2 Flink CDC vs Debezium + Kafka Connect

Code

Option A: Debezium + Kafka Connect + Flink
┌─────┐    ┌──────────┐    ┌───────┐    ┌───────┐
│ DB  │───▶│Debezium  │───▶│ Kafka │───▶│ Flink │───▶ Iceberg
│     │    │(Connect) │    │       │    │       │
└─────┘    └──────────┘    └───────┘    └───────┘

Option B: Flink CDC (Direct)
┌─────┐    ┌──────────────────────────────────┐
│ DB  │───▶│ Flink CDC                         │───▶ Iceberg
│     │    │ (Built-in Debezium)                │
└─────┘    └──────────────────────────────────┘

Dimension	Option A	Option B (Flink CDC)
Component count	3	1
Latency	Seconds	Sub-second
Exactly-once	Difficult	Native support
Operational complexity	High	Low
Data transformation	Separate Flink job	Inline processing

Conclusion: For scenarios requiring data transformation (coomia-dip's primary use case), Flink CDC direct connection is the superior option.

#2. Practice 2: Optimize Full Snapshot Strategy

#2.1 The Snapshot Phase Challenge

When a CDC job starts for the first time, it must perform a full snapshot of the source table. For large tables (hundreds of millions of rows), this can take hours and generate massive network and disk I/O.

#2.2 Chunk-Based Snapshot

Java

// data-Layer/flink-cdc/ChunkedSnapshotConfig.java
MySqlSource<String> source = MySqlSource.<String>builder()
    .hostname("mysql-source")
    .port(3306)
    .databaseList("business_db")
    .tableList("business_db.orders")
    .deserializer(new JsonDebeziumDeserializationSchema())
    // Chunk snapshot configuration
    .splitSize(8096)                    // 8096 rows per chunk
    .splitMetaGroupSize(1000)           // Metadata group size
    .fetchSize(1024)                    // Rows per fetch
    .connectTimeout(Duration.ofSeconds(30))
    .startupOptions(StartupOptions.initial())  // Full + incremental
    .build();

#2.3 Incremental Snapshot (Lock-Free)

Flink CDC 2.0+ introduced an incremental snapshot algorithm that ensures snapshot consistency without global locks:

Code

Phase 1: Parallel Chunk Reading
  ┌─────────┐  ┌─────────┐  ┌─────────┐
  │Chunk 1  │  │Chunk 2  │  │Chunk 3  │  ← Parallel read
  │[1-8096] │  │[8097-   │  │[16193-  │
  │         │  │  16192] │  │  24288] │
  └────┬────┘  └────┬────┘  └────┬────┘
       │            │            │
       ▼            ▼            ▼
Phase 2: Binlog Catchup
  Read Binlog increments during snapshot to fix concurrently modified rows

Phase 3: Switch to Pure Incremental
  Start continuous Binlog consumption from the marked position

#2.4 Skip Snapshot (Incremental Only)

Java

// Read only incremental changes, skip full snapshot
MySqlSource<String> source = MySqlSource.<String>builder()
    .startupOptions(StartupOptions.latest())
    .build();

Use case: The target already has historical data (e.g., from a batch import), and only ongoing changes need synchronization.

#3. Practice 3: Multi-Table CDC Parallel Management

#3.1 Single-Job Multi-Table Mode

Java

// data-Layer/flink-cdc/MultiTableCDC.java
MySqlSource<String> source = MySqlSource.<String>builder()
    .hostname("mysql-source")
    .port(3306)
    .databaseList("business_db")
    .tableList(
        "business_db.orders",
        "business_db.customers",
        "business_db.products",
        "business_db.order_items"
    )
    .deserializer(new JsonDebeziumDeserializationSchema())
    .build();

DataStream<String> cdcStream = env.fromSource(
    source, WatermarkStrategy.noWatermarks(), "MySQL CDC Multi-Table"
);

// Split by table using side outputs
OutputTag<String> ordersTag = new OutputTag<>("orders") {};
OutputTag<String> customersTag = new OutputTag<>("customers") {};

SingleOutputStreamOperator<String> mainStream = cdcStream
    .process(new TableRouter(ordersTag, customersTag));

mainStream.getSideOutput(ordersTag).sinkTo(ordersSink);
mainStream.getSideOutput(customersTag).sinkTo(customersSink);

#3.2 Resource Allocation Guidelines

Table Count	Recommended Parallelism	TaskManager Memory	Notes
1-5	4	4GB	Small scale
5-20	8	8GB	Medium scale
20-50	16	16GB	Large scale
50+	32+	32GB+	Consider splitting jobs

#4. Practice 4: Handle Schema Evolution

#4.1 Source Schema Changes Challenge

In production, source database schemas change frequently. CDC jobs must handle these changes gracefully.

#4.2 Compatible Schema Evolution Handler

Java

// data-Layer/flink-cdc/SchemaEvolutionHandler.java
public class SchemaEvolutionHandler extends ProcessFunction<String, RowData> {

    @Override
    public void processElement(String value, Context ctx,
                               Collector<RowData> out) {
        JsonNode record = objectMapper.readTree(value);
        JsonNode schema = record.get("schema");

        if (isSchemaChanged(schema)) {
            SchemaChangeEvent event = parseSchemaChange(schema);

            switch (event.getType()) {
                case ADD_COLUMN:
                    handleAddColumn(event);
                    break;
                case DROP_COLUMN:
                    handleDropColumn(event);
                    break;
                case ALTER_COLUMN:
                    handleAlterColumn(event);
                    break;
                case RENAME_COLUMN:
                    handleRenameColumn(event);
                    break;
            }
        }

        out.collect(convertToRowData(record));
    }
}

#4.3 Flink CDC 3.0 Schema Evolution

YAML

# Native schema evolution support in Flink CDC 3.0
source:
  type: mysql
  hostname: mysql-source
  port: 3306
  tables: business_db.*
  schema-change-behavior: EVOLVE  # Auto-evolve

sink:
  type: iceberg
  catalog-name: nessie
  catalog-type: rest
  uri: http://nessie-server:19120/api/v2

pipeline:
  schema-evolution:
    enabled: true
    allowed-types:
      - ADD_COLUMN
      - RENAME_COLUMN
      # - DROP_COLUMN  (dangerous — do not auto-execute)

#5. Practice 5: Guarantee Exactly-Once Semantics

#5.1 Three Conditions for End-to-End Exactly-Once

Source: Flink CDC checkpoints record Binlog positions
Processing: Flink checkpoints guarantee internal state consistency
Sink: Target supports idempotent writes or transactional writes

#5.2 Iceberg Sink Exactly-Once

Java

// data-Layer/flink-cdc/ExactlyOnceSink.java
FlinkSink.forRowData(cdcStream)
    .tableLoader(tableLoader)
    .equalityFieldColumns(List.of("order_id"))
    .upsert(true)
    .build();

// Checkpoint configuration
env.enableCheckpointing(60000);
env.getCheckpointConfig().setCheckpointingMode(
    CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
env.getCheckpointConfig().setCheckpointTimeout(600000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);

#5.3 Doris Sink Exactly-Once

Java

// Using Doris Stream Load 2PC protocol
DorisSink.<String>builder()
    .setDorisExecutionOptions(DorisExecutionOptions.builder()
        .setLabelPrefix("flink-cdc-orders")
        .enable2PC()                    // Enable two-phase commit
        .setBufferSize(1024 * 1024)    // 1MB buffer
        .setBufferCount(3)
        .setMaxRetries(3)
        .build())
    .setDorisOptions(DorisOptions.builder()
        .setFenodes("doris-fe:8030")
        .setTableIdentifier("ontology_db.orders")
        .setUsername("root")
        .setPassword("")
        .build())
    .setSerializer(new JsonDebeziumSchemaSerializer(dorisOptions))
    .build();

#6. Practice 6: Watermarks and Late Data Handling

#6.1 Watermark Strategy for CDC

Java

// data-Layer/flink-cdc/WatermarkConfig.java
WatermarkStrategy<CdcEvent> strategy = WatermarkStrategy
    .<CdcEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
    .withTimestampAssigner((event, timestamp) ->
        event.getSourceTimestamp())  // Use source DB timestamp
    .withIdleness(Duration.ofMinutes(1));  // Idle partition timeout

#6.2 Late Data Side Output

Java

OutputTag<CdcEvent> lateDataTag = new OutputTag<>("late-data") {};

SingleOutputStreamOperator<AggregatedResult> result = cdcStream
    .keyBy(event -> event.getTenantId())
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .allowedLateness(Time.minutes(10))
    .sideOutputLateData(lateDataTag)
    .aggregate(new MetricsAggregator());

// Process late data separately
result.getSideOutput(lateDataTag).sinkTo(lateDataSink);

#7. Practice 7: Performance Tuning for Large Tables

#7.1 Parallelism Tuning

Java

env.setParallelism(8);  // Global parallelism

// Source parallelism (snapshot + Binlog reading)
DataStreamSource<String> source = env.fromSource(
    mySqlSource,
    WatermarkStrategy.noWatermarks(),
    "MySQL CDC"
).setParallelism(4);

// Processing parallelism
DataStream<RowData> processed = source
    .map(new CdcMapper()).setParallelism(8);

// Sink parallelism
processed.sinkTo(icebergSink).setParallelism(4);

#7.2 Network Buffer Optimization

YAML

# flink-conf.yaml
taskmanager.network.memory.fraction: 0.15
taskmanager.network.memory.min: 256mb
taskmanager.network.memory.max: 1gb
taskmanager.network.memory.buffers-per-channel: 4
taskmanager.network.memory.floating-buffers-per-gate: 16

#7.3 RocksDB State Backend Tuning

YAML

state.backend: rocksdb
state.backend.rocksdb.memory.managed: true
state.backend.rocksdb.memory.fixed-per-slot: 256mb
state.backend.rocksdb.block.cache-size: 128mb
state.backend.rocksdb.writebuffer.size: 64mb
state.backend.rocksdb.writebuffer.count: 3

#8. Practice 8: Failure Recovery

#8.1 Checkpoint Recovery

Bash

# Resume job from the latest checkpoint
flink run -s hdfs://checkpoint-path/chk-42 \
  -c com.onto.flink.CDCPipeline \
  flink-cdc-pipeline.jar

#8.2 Savepoint Management

Bash

# Create savepoint (for planned downtime)
flink savepoint <job-id> hdfs://savepoints/cdc-orders

# Restore from savepoint (after code upgrade)
flink run -s hdfs://savepoints/cdc-orders \
  --allowNonRestoredState \
  flink-cdc-pipeline-v2.jar

#8.3 Binlog Position Loss Recovery

Java

// When checkpoint is unavailable, restore from a specific Binlog position
MySqlSource<String> source = MySqlSource.<String>builder()
    .startupOptions(StartupOptions.specificOffset(
        "mysql-bin.000003", 154L))
    .build();

#9. Practice 9: Monitoring and Alerting

#9.1 Critical Monitoring Metrics

Metric	Alert Threshold	Description
`sourceEventTimeLag`	> 30s	Source-to-Flink latency
`currentFetchEventTimeLag`	> 60s	Current consumption lag
`numRecordsInPerSecond`	< minimum threshold	Input throughput
`numRecordsOutPerSecond`	< minimum threshold	Output throughput
`checkpointDuration`	> 5min	Checkpoint duration
`checkpointFailureCount`	> 0	Checkpoint failure count
`lastCheckpointSize`	> 10GB	Checkpoint size
`numberOfFailedCheckpoints`	> 3 (consecutive)	Consecutive failures

#9.2 Prometheus + Grafana Integration

YAML

# deployment-Layer/monitoring/prometheus/flink-cdc-alerts.yml
groups:
  - name: flink-cdc-alerts
    rules:
      - alert: CDCLagHigh
        expr: >
          flink_taskmanager_job_task_operator_sourceEventTimeLag > 30000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "CDC lag exceeds 30 seconds"

      - alert: CDCThroughputDrop
        expr: >
          rate(flink_taskmanager_job_task_numRecordsInPerSecond[5m]) < 100
        for: 10m
        labels:
          severity: critical
        annotations:
          summary: "CDC throughput dropped below 100 records/s"

      - alert: CDCCheckpointFailing
        expr: flink_jobmanager_job_numberOfFailedCheckpoints > 3
        for: 15m
        labels:
          severity: critical
        annotations:
          summary: "CDC checkpoint continuously failing"

#10. Practice 10: Flink CDC 3.0 Pipeline Mode

#10.1 Declarative Pipeline

Flink CDC 3.0 introduces a declarative pipeline mode — no Java code required:

YAML

# flink-cdc-pipeline.yaml
source:
  type: mysql
  hostname: mysql-source
  port: 3306
  username: cdc_user
  password: ${MYSQL_CDC_PASSWORD}
  tables: business_db.\.*
  server-id: 5400-5404
  server-time-zone: Asia/Shanghai

sink:
  type: doris
  fenodes: doris-fe:8030
  username: root
  password: ""
  table.create.properties.light_schema_change: true
  table.create.properties.replication_num: 1

transform:
  - source-table: business_db.orders
    projection: order_id, customer_id, amount, status, created_at
    filter: amount > 0
    description: "Filter valid orders"

  - source-table: business_db.customers
    projection: customer_id, name, email, phone
    filter: status = 'ACTIVE'
    description: "Active customers only"

route:
  - source-table: business_db.orders
    sink-table: ontology_db.orders
  - source-table: business_db.customers
    sink-table: ontology_db.customers

pipeline:
  name: coomia-dip-cdc-pipeline
  parallelism: 4
  schema-change-behavior: EVOLVE

#10.2 Execute Pipeline

Bash

# Submit CDC pipeline job
bin/flink-cdc.sh flink-cdc-pipeline.yaml

# Or via Flink REST API
curl -X POST http://flink-jobmanager:8081/jars/flink-cdc.jar/run \
  -d '{"programArgs": "--pipeline flink-cdc-pipeline.yaml"}'

#10.3 Full-Database Sync

YAML

# Full-database sync — one-click MySQL to Doris synchronization
source:
  type: mysql
  hostname: mysql-source
  port: 3306
  tables: business_db.\.*

route:
  - source-table: business_db.\.*
    sink-table: ontology_db.\.*

pipeline:
  name: full-database-sync
  parallelism: 8
  schema-change-behavior: EVOLVE

#Performance Benchmarks

Scenario	Table Size	Snapshot Time	Incremental Latency	Throughput
Single table	10M rows	3 min	< 1s	50K events/s
Single table	100M rows	25 min	< 1s	50K events/s
10 tables parallel	10M rows each	8 min	< 2s	200K events/s
Full DB sync	50 tables / 500M rows	2 hours	< 3s	300K events/s

#Key Takeaways

Flink CDC direct connection beats Debezium + Kafka: Fewer components, lower latency, native exactly-once support. For scenarios requiring data transformation — coomia-dip's primary use case — Flink CDC's inline processing capability is the decisive advantage.
Incremental snapshots are critical for large-table CDC: Flink CDC 2.0+'s lock-free incremental snapshot algorithm eliminates the locking impact on source databases during the full-snapshot phase, making CDC safe for production large tables.
Flink CDC 3.0's declarative pipelines dramatically lower the barrier: No Java code required — YAML configuration enables full-database sync, schema evolution, and data transformation. This is the future direction for coomia-dip's data integration layer.

#Next Article

S8-07: Flink State Management Deep Dive — A deep exploration of Flink state backend selection (HashMapStateBackend vs RocksDBStateBackend), state TTL management, checkpoint tuning, and operational best practices for large-state jobs.

Tags: #flink-cdc #cdc #debezium #real-time-sync #exactly-once #schema-evolution #coomia-dip #Layer-c