3. 环境搭建与快速入门：Flink环境安装、MySQL开启Binlog、Flink CDC Maven依赖、第一个DataStream CDC程序

好，咱们正式开始动手了。这一章的目标很明确——把环境搭起来，跑通第一个 CDC 程序。我见过不少同学在环境上卡一整天，其实说白了就几个关键点：Flink 版本别选错、MySQL Binlog 要开对、依赖别冲突。咱们一个一个来。

3.1 Flink 环境安装：别纠结，用 Standalone 模式就够了

Flink 的部署模式有好几种，YARN、Kubernetes、Standalone。对于学习和本地开发，我个人习惯用 Standalone 模式，简单直接，一条命令启动。

第一步：下载和解压

去 Flink 官网下载稳定版本。我建议用 1.17.x 或更新的版本，CDC 兼容性更好。

wget https://dlcdn.apache.org/flink/flink-1.17.2/flink-1.17.2-bin-scala_2.12.tgz
tar -xzf flink-1.17.2-bin-scala_2.12.tgz
cd flink-1.17.2

第二步：启动集群

./bin/start-cluster.sh

启动后访问 http://localhost:8081，看到 Flink Web UI 就说明成功了。

小提示： 如果你在 Mac 上遇到端口冲突，可以修改 conf/flink-conf.yaml 里的 rest.port 和 jobmanager.rpc.port。

第三步：验证

./bin/flink run examples/streaming/WordCount.jar

看到控制台输出单词计数，环境就稳了。

3.2 MySQL 开启 Binlog：CDC 的数据源头

CDC 要捕获 MySQL 的变更，前提是 MySQL 开启了 Binlog。我记得第一次做 CDC 项目时，DBA 没开 Binlog，我排查了半天才发现……嗯，这个坑咱们提前避开。

检查当前 Binlog 状态

mysql> SHOW VARIABLES LIKE 'log_bin';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| log_bin       | OFF   |
+---------------+-------+

如果是 OFF，需要修改 MySQL 配置文件（my.cnf 或 my.ini）。

开启 Binlog 配置

[mysqld]
server-id = 1
log_bin = mysql-bin
binlog_format = ROW
binlog_row_image = FULL
expire_logs_days = 7
注意： binlog_format 必须设置为 ROW，Flink CDC 只支持 ROW 格式。STATEMENT 和 MIXED 格式会导致数据不一致。

重启 MySQL 并验证

mysql> SHOW MASTER STATUS;
+------------------+----------+--------------+------------------+-------------------+
| File             | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set |
+------------------+----------+--------------+------------------+-------------------+
| mysql-bin.000001 |      154 |              |                  |                   |
+------------------+----------+--------------+------------------+-------------------+

看到 File 和 Position 有值，Binlog 就正常工作了。

3.3 Flink CDC Maven 依赖：版本匹配是关键

依赖配置是新手最容易踩坑的地方。我曾经见过有人把 Flink 1.14 和 CDC 2.4 混用，结果跑起来全是序列化异常。这里我直接给出一套经过验证的依赖组合。

核心依赖（基于 Flink 1.17 + CDC 2.4）

<properties>
    <flink.version>1.17.2</flink.version>
    <cdc.version>2.4.0</cdc.version>
    <scala.binary.version>2.12</scala.binary.version>
</properties>

<dependencies>
    <!-- Flink 核心 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <!-- Flink CDC -->
    <dependency>
        <groupId>com.ververica</groupId>
        <artifactId>flink-connector-mysql-cdc</artifactId>
        <version>${cdc.version}</version>
    </dependency>

    <!-- MySQL JDBC 驱动 -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.33</version>
    </dependency>
</dependencies>
版本匹配速查表：
Flink 版本CDC 推荐版本Scala 版本
1.14.x2.2.x2.12
1.15.x2.3.x2.12
1.16.x2.3.x / 2.4.x2.12
1.17.x2.4.x2.12

3.4 第一个 DataStream CDC 程序：从 MySQL 读取变更

环境搭好了，依赖配好了，咱们写个最简单的程序——监听 MySQL 的一张表，把增删改查打印出来。

准备测试表

CREATE DATABASE flink_cdc_demo;
USE flink_cdc_demo;

CREATE TABLE users (
    id INT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(50),
    email VARCHAR(100),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

INSERT INTO users(name, email) VALUES('Alice', 'alice@example.com');
INSERT INTO users(name, email) VALUES('Bob', 'bob@example.com');

编写 CDC 程序

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;

public class MySqlCdcDemo {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 2. 配置 MySQL CDC Source
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname("localhost")
            .port(3306)
            .databaseList("flink_cdc_demo")
            .tableList("flink_cdc_demo.users")
            .username("root")
            .password("your_password")
            .deserializer(new JsonDebeziumDeserializationSchema())
            .build();

        // 3. 读取数据流
        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC Source")
            .print();

        // 4. 执行
        env.execute("Flink CDC Demo");
    }
}

运行程序

在 MySQL 里执行一些变更操作：

INSERT INTO users(name, email) VALUES('Charlie', 'charlie@example.com');
UPDATE users SET email = 'alice_new@example.com' WHERE id = 1;
DELETE FROM users WHERE id = 2;

控制台会输出类似这样的 JSON：

{
  "before": null,
  "after": {
    "id": 3,
    "name": "Charlie",
    "email": "charlie@example.com",
    "created_at": 1695000000000
  },
  "op": "c"
}

{
  "before": {
    "id": 1,
    "name": "Alice",
    "email": "alice@example.com"
  },
  "after": {
    "id": 1,
    "name": "Alice",
    "email": "alice_new@example.com"
  },
  "op": "u"
}

{
  "before": {
    "id": 2,
    "name": "Bob",
    "email": "bob@example.com"
  },
  "after": null,
  "op": "d"
}

op 字段表示操作类型：c 是新增，u 是更新，d 是删除。你看，MySQL 里的每一次变更，都实时地流到了 Flink 里。

3.5 核心流程一览

下面这张图把整个 CDC 数据流串起来了，你一看就明白。

Flink CDC 数据流架构 MySQL 数据库 Binlog 日志 Flink CDC Connector DataStream API 写入 解析捕获 Source 流 说明： MySQL 将变更写入 Binlog 日志文件，Flink CDC Connector 模拟 MySQL Slave 读取 Binlog， 解析为变更事件流，通过 DataStream API 供下游消费。

3.6 避坑指南

  • MySQL 用户权限： CDC 用户需要 SELECT、REPLICATION SLAVE、REPLICATION CLIENT 权限。我曾经因为漏了 REPLICATION SLAVE，程序一直连不上 Binlog。
  • 并行度设置： MySQL CDC Source 的并行度不能超过表的数量。如果你只监听一张表，并行度设为 1 最稳。
  • 时区问题： Flink CDC 默认使用 UTC 时区。如果你的 MySQL 是东八区，记得在连接参数里加 ?serverTimezone=Asia/Shanghai。
  • 大表初始化： 第一次启动 CDC 时，Flink 会先做一次全量快照，再切换到增量 Binlog。如果表很大，建议先做一次 SELECT COUNT(*) 评估数据量。

好了，环境搭好了，第一个程序也跑通了。你现在应该能感受到 CDC 的魅力——数据库的每一次变更，都像流水一样实时流到你的面前。后面咱们会在这个基础上，做更复杂的处理和落地。

← 上一章 📖 返回目录 下一章 →