Build Real-Time MySQL CDC Pipelines with Flink 1.19 and SpringBoot

1. Introduction

Flink CDC (Change Data Capture) is a CDC framework based on database logs that integrates with the Flink compute engine to provide efficient real‑time data integration for massive data streams. It monitors database changes, extracts them, and makes them available for further processing and analysis.

Supported Databases

MySQL 5.6, 5.7, 8.0.x

RDS MySQL 5.6, 5.7, 8.0.x

PolarDB MySQL 5.6, 5.7, 8.0.x

Aurora MySQL 5.6, 5.7, 8.0.x

MariaDB 10.x

PolarDB X 2.0.1

2. Practical Example

2.1 Enable MySQL Binlog

<code>[mysqld]
server-id=1
# row‑level format
binlog-format=Row
# binlog file prefix
log-bin=mysql-bin
# databases to capture
binlog_do_db=testjpa</code>

In addition to enabling binlog, the Flink CDC user must have SELECT, REPLICATION SLAVE, REPLICATION CLIENT, SHOW VIEW, etc., privileges to read data and metadata.

2.2 Dependency Management

<code>&lt;properties&gt;
  &lt;flink.version&gt;1.19.0&lt;/flink.version&gt;
&lt;/properties&gt;
&lt;dependency&gt;
  &lt;groupId&gt;org.apache.flink&lt;/groupId&gt;
  &lt;artifactId&gt;flink-connector-base&lt;/artifactId&gt;
  &lt;version&gt;${flink.version}&lt;/version&gt;
&lt;/dependency&gt;
&lt;dependency&gt;
  &lt;groupId&gt;com.ververica&lt;/groupId&gt;
  &lt;artifactId&gt;flink-sql-connector-mysql-cdc&lt;/artifactId&gt;
  &lt;version&gt;3.0.1&lt;/version&gt;
&lt;/dependency&gt;
... (additional Flink dependencies) ...</code>

2.3 Code Implementation

<code>@Component
public class MonitorMySQLCDC implements InitializingBean {
    public static final LinkedBlockingQueue&lt;Map&lt;String, Object&gt;&gt; queue = new LinkedBlockingQueue&lt;&gt;();
    private final StringRedisTemplate stringRedisTemplate;
    private final String PREFIX = "users:";
    private final CustomSink customSink;
    public MonitorMySQLCDC(CustomSink customSink, StringRedisTemplate stringRedisTemplate) {
        this.customSink = customSink;
        this.stringRedisTemplate = stringRedisTemplate;
    }
    @Override
    public void afterPropertiesSet() throws Exception {
        new Thread(() -> {
            while (true) {
                try {
                    Map&lt;String, Object&gt; result = queue.take();
                    this.doAction(result);
                } catch (Exception e) { e.printStackTrace(); }
            }
        }).start();
        Properties jdbcProperties = new Properties();
        jdbcProperties.setProperty("useSSL", "false");
        MySqlSource&lt;String&gt; source = MySqlSource.&lt;String&gt;builder()
            .hostname("127.0.0.1")
            .port(3306)
            .databaseList("testjpa")
            .tableList("testjpa.users")
            .username("root")
            .password("123123")
            .jdbcProperties(jdbcProperties)
            .includeSchemaChanges(true)
            .deserializer(new JsonDebeziumDeserializationSchema(true))
            .startupOptions(StartupOptions.initial())
            .build();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(6000);
        env.setParallelism(4);
        env.fromSource(source, WatermarkStrategy.noWatermarks(), "MySQL")
            .addSink(this.customSink);
        env.execute();
    }
    @SuppressWarnings("unchecked")
    private void doAction(Map&lt;String, Object&gt; result) throws Exception {
        Map&lt;String, Object&gt; payload = (Map&lt;String, Object&gt;) result.get("payload");
        String op = (String) payload.get("op");
        switch (op) {
            case "u", "c" -> {
                Map&lt;String, Object&gt; after = (Map&lt;String, Object&gt;) payload.get("after");
                String id = after.get("id").toString();
                System.out.printf("操作：%s, ID: %s%n", op, id);
                stringRedisTemplate.opsForValue().set(PREFIX + id, new ObjectMapper().writeValueAsString(after));
            }
            case "d" -> {
                Map&lt;String, Object&gt; before = (Map&lt;String, Object&gt;) payload.get("before");
                String id = before.get("id").toString();
                stringRedisTemplate.delete(PREFIX + id);
            }
        }
    }
}
</code>

Startup Modes

initial (default): take an initial snapshot on first start, then continue reading the latest binlog.

earliest-offset : skip snapshot, start from the earliest available binlog position.

latest-offset : do not snapshot; start reading from the end of the binlog, capturing only changes after the connector starts.

specific-offset : skip snapshot and start from a user‑specified binlog file and position (or GTID set).

timestamp : skip snapshot and start from a given timestamp.

Data Processing Sink

<code>@Component
public class CustomSink extends RichSinkFunction&lt;String&gt; {
    private ObjectMapper mapper = new ObjectMapper();
    @Override
    public void invoke(String value, Context context) throws Exception {
        System.out.printf("数据发生变化: %s%n", value);
        TypeReference&lt;Map&lt;String, Object&gt;&gt; typeRef = new TypeReference&lt;&gt;() {};
        Map&lt;String, Object&gt; result = mapper.readValue(value, typeRef);
        Map&lt;String, Object&gt; payload = (Map&lt;String, Object&gt;) result.get("payload");
        String op = (String) payload.get("op");
        if (!"r".equals(op)) {
            MonitorMySQLCDC.queue.put(result);
        }
    }
}
</code>

Web Monitoring Page

Add the Flink web runtime dependency and configure the REST port.

<code>&lt;dependency&gt;
  &lt;groupId&gt;org.apache.flink&lt;/groupId&gt;
  &lt;artifactId&gt;flink-runtime-web&lt;/artifactId&gt;
  &lt;version&gt;${flink.version}&lt;/version&gt;
&lt;/dependency&gt;
</code>

<code>Configuration config = new Configuration();
config.set(RestOptions.PORT, 9090);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
</code>

The web UI listens on port 9090.

This article has covered the complete implementation of real‑time MySQL CDC to Redis using Flink.