3、环境搭建与快速入门：Flink与Flink CDC环境安装配置、MySQL Binlog开启、第一个Flink CDC程序

好，咱们直接进入正题。这一章说白了就是「动手干」。很多同学学 Flink CDC 卡在第一关——环境装不上，或者装上了跑不起来。我当年第一次搭环境时也踩了不少坑，尤其是 MySQL Binlog 那块的权限问题，折腾了大半天。今天我把这些经验都揉碎了讲给你听。

3.1 环境准备：我们需要什么？

先列个清单，看看咱们需要哪些组件。我个人习惯用 Docker 来跑 MySQL，省得污染本地环境。但 Flink 集群我建议直接本地解压跑，调试起来更直观。

3.2 安装 Flink 本地集群

这一步其实很简单。下载、解压、启动，三步走。

# 1. 下载 Flink（以 1.15.2 为例）
wget https://archive.apache.org/dist/flink/flink-1.15.2/flink-1.15.2-bin-scala_2.12.tgz

# 2. 解压
tar -xzf flink-1.15.2-bin-scala_2.12.tgz
cd flink-1.15.2

# 3. 启动本地集群
./bin/start-cluster.sh

启动后，打开浏览器访问 http://localhost:8081，看到 Flink Web UI 就说明成功了。嗯，这里要注意——如果你 8081 端口被占了，可以去 conf/flink-conf.yaml 里改 rest.port。

3.3 配置 MySQL：开启 Binlog

这是整个环节里最容易出问题的地方。我曾经在客户现场遇到过 MySQL 8.0 默认没开 Binlog，导致 CDC 跑不起来，排查了半天才发现是配置问题。

首先，确认你的 MySQL 是否开启了 Binlog：

-- 登录 MySQL
mysql -u root -p

-- 查看 Binlog 状态
SHOW VARIABLES LIKE 'log_bin';

如果 Value 是 OFF，那就需要修改 MySQL 配置文件。找到 my.cnf（Linux 通常在 /etc/mysql/my.cnf 或 /etc/my.cnf），添加以下内容：

[mysqld]
# 必须开启
log_bin=mysql-bin
# 必须设置为 ROW
binlog_format=ROW
# 设置 server_id，集群中每个节点唯一
server_id=1
# 可选：指定要监听的数据库，不指定则监听所有
# binlog_do_db=your_database

改完后重启 MySQL：

sudo systemctl restart mysql
# 或者
sudo service mysql restart

重启后再次检查，确认 Binlog 已开启：

SHOW VARIABLES LIKE 'log_bin';
SHOW VARIABLES LIKE 'binlog_format';

3.4 创建 CDC 专用用户

安全起见，别直接用 root。创建一个专门给 Flink CDC 用的用户：

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

这三个权限的含义：

• SELECT：读取表结构，做快照
• REPLICATION SLAVE：读取 Binlog 日志
• REPLICATION CLIENT：查看主库状态

3.5 准备测试数据

创建一个测试库和表，方便咱们后面跑程序：

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 ('张三', 'zhangsan@example.com');
INSERT INTO users(name, email) VALUES ('李四', 'lisi@example.com');

3.6 第一个 Flink CDC 程序

好，重头戏来了。咱们写一个最简单的 Flink CDC 程序，读取 MySQL 的变更数据并打印到控制台。

首先，创建一个 Maven 项目，在 pom.xml 里添加依赖：

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.15.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>1.15.2</version>
    </dependency>
    <dependency>
        <groupId>com.ververica</groupId>
        <artifactId>flink-connector-mysql-cdc</artifactId>
        <version>2.3.0</version>
    </dependency>
</dependencies>

然后写主程序：

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

public class MySqlCdcDemo {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 开启 checkpoint，保证 exactly-once
        env.enableCheckpointing(5000);

        // 2. 创建 MySQL CDC Source
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname("localhost")
            .port(3306)
            .databaseList("flink_cdc_demo")   // 监听的数据库
            .tableList("flink_cdc_demo.users") // 监听的表
            .username("cdc_user")
            .password("cdc_password")
            .deserializer(new JsonDebeziumDeserializationSchema()) // 输出 JSON
            .build();

        // 3. 接入 Source 并打印
        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC Source")
           .print();

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

代码写完了，怎么跑？有两种方式：

• 直接在 IDE 里跑：适合开发和调试
• 打包提交到 Flink 集群：适合生产

咱们先直接在 IDE 里跑。运行 main 方法，你会看到控制台输出类似这样的 JSON：

{
  "before": null,
  "after": {
    "id": 1,
    "name": "张三",
    "email": "zhangsan@example.com",
    "created_at": 1695000000000
  },
  "source": {
    "db": "flink_cdc_demo",
    "table": "users"
  },
  "op": "r"
}

这个 op 字段表示操作类型：

• r：读取（快照阶段）
• c：创建（insert）
• u：更新（update）
• d：删除（delete）

现在，去 MySQL 里执行一条插入语句：

INSERT INTO users(name, email) VALUES ('王五', 'wangwu@example.com');

你会立刻看到控制台输出一条新的 JSON，op 变成了 c。这就是实时同步的效果。

3.7 本章知识体系总览

下面这张图帮你梳理了本章的核心流程和组件关系：

3.8 常见问题与避坑

最后，我把这些年遇到的典型问题列出来，你遇到了可以直接对照排查：

好了，环境搭好了，第一个程序也跑通了。你现在应该能感受到 Flink CDC 的魅力——几行代码就能实现数据库的实时变更捕获。后面的章节我们会深入解析它的内部原理和高级用法。