“

Series: S12 Developer Tutorials · Article 15 | Level: Intermediate | Reading Time: 15 min

Flink CDC Real-Time Data Sync Guide

#Introduction

In enterprise data platforms, data synchronization is one of the most fundamental yet critical capabilities. Traditional batch ETL is reliable but cannot meet real-time decision-making needs. Flink CDC (Change Data Capture) enables you to capture database change events in real time, streaming them into coomia-dip's data layer so that every row change in your business database is reflected in the analytics platform within milliseconds.

This tutorial walks you through CDC fundamentals, MySQL binlog configuration, Flink cluster deployment, and implementing a real-time sync pipeline from MySQL to Apache Doris.

#1. CDC Fundamentals

#1.1 What is Change Data Capture

Change Data Capture is a technique that monitors database transaction logs to capture data changes. Compared to periodic full-table pulls, CDC offers three core advantages:

Real-time: Changes captured within milliseconds, not waiting for schedule cycles
Low overhead: Only changed data transmitted; no full table scans
Completeness: Captures all changes including intermediate states, not just final snapshots

The workflow: source database transaction logs are continuously consumed by Flink CDC Connector, parsed into structured change events (Changelog Stream), processed through Flink's stream processing for cleaning, transformation and routing, then written to the target system (Apache Doris in coomia-dip).

#1.2 Flink CDC 3.x Pipeline Architecture

Flink CDC 3.x introduces a declarative Pipeline architecture that drastically simplifies sync task configuration:

YAML

source:
  type: mysql
  hostname: mysql-source
  port: 3306
  username: cdc_user
  password: ${MYSQL_CDC_PASSWORD}
  tables: erp_db.orders, erp_db.order_items, erp_db.customers

sink:
  type: doris
  fenodes: doris-fe:8030
  username: root
  password: ${DORIS_PASSWORD}

pipeline:
  name: erp-to-doris-sync
  parallelism: 4
  schema.change.behavior: evolve

#1.3 CDC's Role in coomia-dip

Within coomia-dip's Layered architecture, CDC is a core capability of the Data Layer (Data Layer). Data flows from business databases through CDC into the Doris ODS layer, is auto-mapped to Ontology objects by the Ontology Mapper, and served through the OQL engine for unified querying.

#2. Environment Setup

#2.1 Enable MySQL Binlog

CDC requires MySQL's binlog in ROW format for complete field-level change information:

INI

[mysqld]
server-id         = 1
log_bin           = mysql-bin
binlog_format     = ROW
binlog_row_image  = FULL
expire_logs_days  = 3
gtid_mode         = ON
enforce_gtid_consistency = ON

Create a dedicated CDC user:

SQL

CREATE USER 'cdc_user'@'%' IDENTIFIED BY 'cdc_secure_password';
GRANT SELECT, RELOAD, SHOW DATABASES,
      REPLICATION SLAVE, REPLICATION CLIENT
    ON *.* TO 'cdc_user'@'%';
FLUSH PRIVILEGES;

#2.2 Deploy Flink Cluster

YAML

version: "3.8"
services:
  jobmanager:
    image: flink:1.18-java11
    ports: ["8081:8081"]
    command: jobmanager
    environment:
      FLINK_PROPERTIES: |
        jobmanager.rpc.address: jobmanager
        state.checkpoints.dir: file:///opt/flink/checkpoints
        execution.checkpointing.interval: 60000

  taskmanager:
    image: flink:1.18-java11
    depends_on: [jobmanager]
    command: taskmanager
    deploy:
      replicas: 2
    environment:
      FLINK_PROPERTIES: |
        jobmanager.rpc.address: jobmanager
        taskmanager.numberOfTaskSlots: 4
        taskmanager.memory.process.size: 4096m

#2.3 Install CDC Connectors

Bash

FLINK_LIB=/opt/flink/lib
wget -P $FLINK_LIB https://repo1.maven.org/.../flink-sql-connector-mysql-cdc-3.0.1.jar
wget -P $FLINK_LIB https://repo1.maven.org/.../flink-doris-connector-1.18-1.6.1.jar

#3. Implementing the Sync Pipeline

#3.1 Flink SQL (Recommended for Beginners)

SQL

SET 'execution.runtime-mode' = 'streaming';

CREATE TABLE orders_source (
    order_id BIGINT, customer_id BIGINT, product_name STRING,
    quantity INT, unit_price DECIMAL(10,2), total_amount DECIMAL(12,2),
    status STRING, created_at TIMESTAMP(3), updated_at TIMESTAMP(3),
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'mysql-source', 'port' = '3306',
    'username' = 'cdc_user', 'password' = 'cdc_secure_password',
    'database-name' = 'erp_db', 'table-name' = 'orders',
    'scan.incremental.snapshot.enabled' = 'true'
);

CREATE TABLE orders_sink (
    order_id BIGINT, customer_id BIGINT, product_name STRING,
    quantity INT, unit_price DECIMAL(10,2), total_amount DECIMAL(12,2),
    status STRING, created_at TIMESTAMP(3), updated_at TIMESTAMP(3),
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'doris', 'fenodes' = 'doris-fe:8030',
    'table.identifier' = 'ods.orders',
    'username' = 'root', 'password' = '',
    'sink.enable-delete' = 'true'
);

INSERT INTO orders_sink SELECT * FROM orders_source;

#3.2 DataStream API (Flexible Customization)

Java

package com.onto.pipeline.cdc;

public class ERPSyncPipeline {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(60_000, CheckpointingMode.EXACTLY_ONCE);
        env.setParallelism(4);

        MySqlSource<String> source = MySqlSource.<String>builder()
            .hostname("mysql-source").port(3306)
            .databaseList("erp_db")
            .tableList("erp_db.orders", "erp_db.order_items")
            .username("cdc_user").password("cdc_secure_password")
            .deserializer(new JsonDebeziumDeserializationSchema())
            .includeSchemaChanges(true).splitSize(8096)
            .build();

        env.fromSource(source, WatermarkStrategy.noWatermarks(), "MySQL CDC")
            .map(new CDCEventTransformer())
            .process(new TableRoutingFunction())
            .sinkTo(DorisSinkBuilder.build());

        env.execute("ERP CDC Pipeline");
    }
}

#4. Ontology Integration

#4.1 Register CDC Data Source

Python

from ontology_sdk import OntoPlatform
platform = OntoPlatform(base_url="http://localhost:8080", token="admin-token")

source = platform.data_sources.register(
    name="erp-mysql-cdc",
    source_type="FLINK_CDC",
    config={
        "connector": "mysql-cdc", "hostname": "mysql-source",
        "port": 3306, "database": "erp_db",
        "tables": ["orders", "order_items", "customers"],
        "sync_mode": "INCREMENTAL",
    },
    target={"type": "DORIS", "database": "ods"},
)

#4.2 Auto-Map to Ontology Objects

Python

platform.ontology.create_mapping(
    source_table="ods.orders",
    object_type="Order",
    field_mappings={
        "order_id":     {"property": "orderId", "type": "STRING", "primary_key": True},
        "customer_id":  {"property": "customerId", "type": "STRING"},
        "total_amount": {"property": "totalAmount", "type": "DOUBLE"},
        "status":       {"property": "status", "type": "STRING"},
    },
    sync_mode="REAL_TIME",
    link_rules=[{
        "link_type": "orderedBy",
        "target_object_type": "Customer",
        "source_field": "customer_id",
        "target_field": "customerId",
    }],
)

#4.3 Query Real-Time Data via OQL

Python

result = platform.oql.execute("""
    SELECT o.orderId, o.totalAmount, c.name AS customerName
    FROM Order o JOIN o.orderedBy c
    WHERE o.totalAmount > 10000
      AND o.createdAt > NOW() - INTERVAL 1 HOUR
    ORDER BY o.totalAmount DESC LIMIT 20
""")

#5. Monitoring & Operations

#5.1 Key Metrics

Metric	Meaning	Alert Threshold
`numRecordsInPerSecond`	Input records/s	Spike > 200% or 0
`numRecordsOutPerSecond`	Output records/s	Gap > 20% vs input
`currentFetchEventTimeLag`	CDC event lag	> 60 seconds
`lastCheckpointDuration`	Checkpoint time	> 30 seconds
`numberOfFailedCheckpoints`	Failed count	> 0

#5.2 Common Issues

Problem	Cause	Solution
Job fails on start	binlog expired	Increase retention, full resync
OOM during snapshot	Table too large	Reduce chunk.size, add memory
Doris write timeout	Stream Load slow	Increase timeout, reduce batch
Schema change crash	DDL not handled	Enable schema evolution
Growing latency	Sink bottleneck	Increase parallelism

#5.3 Disaster Recovery

Bash

# Create savepoint
curl -X POST http://flink:8081/jobs/{JOB_ID}/savepoints \
  -d '{"cancel-job": false, "target-directory": "file:///opt/flink/savepoints"}'

# Recover from savepoint
./bin/flink run -s file:///opt/flink/savepoints/savepoint-xxx \
  -c com.onto.pipeline.cdc.ERPSyncPipeline pipeline.jar

#6. Production Best Practices

#6.1 Capacity Planning

Source Size	Change TPS	Parallelism	TM Memory
< 1M rows	< 100	2	2 GB
1-10M	100-1000	4	4 GB
10-100M	1000-5000	8	8 GB
> 100M	> 5000	16+	16 GB+

#6.2 Pre-Launch Checklist

MySQL binlog = ROW with FULL row image
binlog retention >= 3 days
CDC user has minimal permissions
Flink checkpoints on persistent storage
Monitoring alerts configured
Doris target schema created
Network connectivity verified
Savepoint recovery tested

#Summary

This tutorial covered real-time data synchronization with Flink CDC: fundamentals, environment setup, Flink SQL and DataStream API implementations, Ontology auto-mapping, monitoring, and production best practices. Flink CDC is the circulatory system of coomia-dip's data layer, enabling real-time data convergence into a unified Ontology model.

Next: [S12-16] Temporal Workflow Orchestration Guide Previous: [S12-14] gRPC Custom Service Development Guide