企业官网建设流程全解析

Flink CDC 3.5.0 从理论到实践 —— 第 3 章 环境准备与版本依赖

课程定位：本系列教程以MySQL 为唯一数据源，Sink 覆盖Doris / Paimon / Kafka三大目标，从原理到生产落地，全链路实战。

版本基线：Flink CDC 3.5.0 + Flink 1.20.x + MySQL 8.0/8.4 + Doris 4.1 + Paimon 1.4.2+ + Kafka 3.x

章节导读

• 3.1 软件版本矩阵与兼容性
• 3.2 MySQL 端配置
• 3.3 Flink 集群准备
• 3.4 依赖获取与部署清单
• 3.5 课堂环境快速拉起（Docker Compose）
• 3.6 环境连通性验证
• 3.7 本章小结

3.1 软件版本矩阵与兼容性

环境准备的第一步是锁定版本矩阵。CDC 链路组件多、依赖关系复杂，版本不匹配是新手最容易踩坑的地方，务必先对齐再动手。

3.1.1 本教程版本基线

3.1.2 连接器兼容性矩阵

Flink CDC 3.5.0 与各 Sink 连接器的兼容关系（关键版本约束）：

注意：

1. Flink CDC 3.x 各版本对 Flink 小版本有严格要求（如 3.5.0 面向 Flink 1.20），混用会导致NoSuchMethodError等运行时错误；
2. flink-doris-connector与 Flink CDC 的 Doris Pipeline 是两个不同的制品：前者用于 Flink SQL 读写 Doris，后者用于整库同步管道，不要混淆。

3.1.3 部署拓扑规划

本教程使用的组件拓扑与端口约定：

┌─────────────┐ 3306 ┌──────────────────────┐ │ MySQL │◄────────│ Flink 1.20 (Standalone) │ │ (数据源) │ Binlog │ JobManager : 8081 │ └─────────────┘ │ TaskManager x N │ └───────┬──────┬──────┬─────┘ │ │ │ ┌──────────▼┐ ┌───▼──────┐ ┌▼───────────┐ │ Doris 4.1│ │ Paimon │ │ Kafka │ │ FE : 8030 │ │ (HDFS/ │ │ Broker:9092│ │ 查询: 9030 │ │ 本地FS) │ │ │ │ BE : 8040 │ │ │ │ │ └───────────┘ └──────────┘ └────────────┘

3.2 MySQL 端配置

MySQL 是整条链路的源头，Binlog 配置与账号权限必须先就位，否则 Flink CDC 作业启动即报错。

3.2.1 前置要求清单

3.2.2 my.cnf 配置示例

[mysqld] # ===== Binlog 配置（CDC 必需）===== server-id = 1 log_bin = mysql-bin binlog_format = ROW binlog_row_image = FULL # Binlog 保留 7 天（按业务调整，全量初始化耗时长的场景建议更长） binlog_expire_logs_seconds = 604800 # ===== GTID（推荐开启，便于按 GTID 恢复位点）===== gtid_mode = ON enforce_gtid_consistency = ON # ===== 连接超时（全量大表初始化时避免连接被回收）===== interactive_timeout = 28800 wait_timeout = 28800

修改后重启 MySQL 生效：

systemctl restart mysqld

验证配置：

SHOWVARIABLESLIKE'log_bin';-- ONSHOWVARIABLESLIKE'binlog_format';-- ROWSHOWVARIABLESLIKE'binlog_row_image';-- FULLSHOWVARIABLESLIKE'gtid_mode';-- ONSHOWVARIABLESLIKE'binlog_expire_logs_seconds';-- 604800

3.2.3 创建 CDC 专用账号

不要使用 root 做 CDC 采集，按最小权限原则创建专用账号：

-- 1. 创建用户（生产建议限制来源 IP）CREATEUSER'cdc_user'@'%'IDENTIFIEDBY'Cdc.2026';-- 2. 授予最小权限GRANTSELECT,SHOWDATABASES,REPLICATIONSLAVE,REPLICATIONCLIENTON*.*TO'cdc_user'@'%';-- 3. 刷新权限FLUSHPRIVILEGES;-- 4. 验证SHOWGRANTSFOR'cdc_user'@'%';

权限说明：

注意：增量快照模式下无需RELOAD权限（旧版本 Debezium 快照需要），但若关闭增量快照则需要补充。

3.2.4 准备业务测试数据

创建一个业务库用于后续章节实战：

CREATEDATABASEappDEFAULTCHARACTERSETutf8mb4;USEapp;CREATETABLEorders(idBIGINTNOTNULLAUTO_INCREMENT,customer_idBIGINTNOTNULL,amountDECIMAL(10,2)NOTNULL,statusVARCHAR(20)NOTNULLDEFAULT'created',create_timeDATETIMENOTNULLDEFAULTCURRENT_TIMESTAMP,update_timeDATETIMENOTNULLDEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMP,PRIMARYKEY(id))ENGINE=InnoDB;INSERTINTOorders(customer_id,amount,status)VALUES(1001,99.90,'paid'),(1002,199.00,'created'),(1003,59.50,'paid');

要点：CDC 表必须有主键，否则无法做增量快照并发切分（无主键表需指定chunk.key-column，见第 2 章）。

3.3 Flink 集群准备

Flink CDC 3.5.0 是运行在 Flink 之上的管道框架，先准备好 Flink 集群。本节以Standalone 模式为主线（最简单、适合学习），并给出 YARN / K8s 的差异说明。

3.3.1 下载与解压

# 下载 Flink 1.20.x（Scala 版本后缀不影响，1.20 起已无 Scala 区分）wgethttps://archive.apache.org/dist/flink/flink-1.20.1/flink-1.20.1-bin-scala_2.12.tgz# 解压到部署目录tar-xzfflink-1.20.1-bin-scala_2.12.tgz-C/opt/apps/ln-s/opt/apps/flink-1.20.1 /opt/apps/flinkexportFLINK_HOME=/opt/apps/flink

3.3.2 config.yaml 核心配置

Flink 1.20 使用新的config.yaml（扁平化格式，替代旧的flink-conf.yaml）：

# ===== 基础 =====jobmanager.rpc.address:localhostjobmanager.bind-host:0.0.0.0rest.bind-address:0.0.0.0rest.port:8081# ===== 内存（按机器规格调整）=====jobmanager.memory.process.size:2048mtaskmanager.memory.process.size:4096mtaskmanager.numberOfTaskSlots:4# ===== Checkpoint（Exactly-Once 前提，见第 2 章）=====execution.checkpointing.interval:60sexecution.checkpointing.mode:EXACTLY_ONCEexecution.checkpointing.timeout:10minexecution.checkpointing.min-pause:30sexecution.checkpointing.externalized-checkpoint-retention:RETAIN_ON_CANCELLATIONstate.checkpoints.dir:file:///opt/apps/flink/checkpointsstate.savepoints.dir:file:///opt/apps/flink/savepoints# ===== 重启策略 =====restart-strategy.type:fixed-delayrestart-strategy.fixed-delay.attempts:3restart-strategy.fixed-delay.delay:10s# ===== 并行度默认值 =====parallelism.default:2

说明：

1. 开发环境 Checkpoint 目录用本地盘（file://）即可；生产建议 HDFS / S3（hdfs:///s3://）；
2. RETAIN_ON_CANCELLATION保证作业取消后 Checkpoint 保留，便于从断点恢复；
3. 若继续使用旧版flink-conf.yaml，Flink 1.20 仍兼容，但推荐迁移到config.yaml。

3.3.3 启动与验证

cd/opt/apps/flink# 启动 Standalone 集群./bin/start-cluster.sh# 验证进程（应有 StandaloneSessionClusterEntrypoint 和 TaskManagerRunner）jps|grep-E"StandaloneSession|TaskManager"# 验证 WebUI（浏览器访问 http://<host>:8081）curl-shttp://localhost:8081/overview

返回类似 JSON 即集群正常：

{"taskmanagers":1,"slots-total":4,"slots-available":4,"jobs-running":0,"jobs-finished":0,"jobs-cancelled":0,"jobs-failed":0,"flink-version":"1.20.1"}

3.3.4 其他部署模式速览

YARN 模式需要先导出 Hadoop 环境变量：

exportHADOOP_CLASSPATH=$(hadoop classpath)# 提交示例./bin/flink run-tyarn-per-job-ccom.xxx.MainApp app.jar

本教程主线使用 Standalone 模式讲解，YARN / K8s 的生产化部署在第 10 章运维篇展开。

3.4 依赖获取与部署清单

Flink CDC 的依赖分两组：Flink CDC 发行包（Pipeline 作业运行时）与Flink lib 目录的连接器 jar（SQL API 作业使用）。

3.4.1 Flink CDC 发行包

# 下载 Flink CDC 3.5.0 二进制发行包wgethttps://archive.apache.org/dist/flink/flink-cdc-3.5.0/flink-cdc-3.5.0-bin.tar.gz# 解压tar-xzfflink-cdc-3.5.0-bin.tar.gz-C/opt/apps/# 目录结构/opt/apps/flink-cdc-3.5.0/ ├── bin/ │ └── flink-cdc.sh# Pipeline 作业提交入口├── lib/ │ ├── flink-cdc-dist-3.5.0.jar │ ├── flink-cdc-pipeline-connector-mysql-3.5.0.jar │ ├── flink-cdc-pipeline-connector-doris-3.5.0.jar │ ├── flink-cdc-pipeline-connector-kafka-3.5.0.jar │ ├── flink-cdc-pipeline-connector-paimon-3.5.0.jar │ └── flink-cdc-runtime-3.5.0.jar └── conf/# Pipeline YAML 存放处（自建）

要点：Pipeline 作业通过bin/flink-cdc.sh提交，它会把自身lib/下的连接器一起加载，MySQL / Doris / Kafka / Paimon 四个 Pipeline 连接器均已内置，开箱即用。

3.4.2 Flink lib 目录依赖

若使用Flink SQL API方式（单表 CDC），需要将对应 jar 放入 Flink 的lib/：

cd$FLINK_HOME/lib# 示例（版本号以实际下载为准）wgethttps://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-mysql-cdc/3.5.0/flink-sql-connector-mysql-cdc-3.5.0.jarwgethttps://repo1.maven.org/maven2/org/apache/paimon/paimon-flink-1.20/1.4.2/paimon-flink-1.20-1.4.2.jarwgethttps://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-kafka/3.4.0-1.20/flink-sql-connector-kafka-3.4.0-1.20.jarwgethttps://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.33/mysql-connector-j-8.0.33.jar# 重启集群使 jar 生效./bin/stop-cluster.sh&&./bin/start-cluster.sh

注意：

1. lib/下不要放同一制品的多个版本，避免类冲突；
2. Paimon jar 同时放入flink-cdc-3.5.0/lib/（Pipeline 方式写 Paimon 时也需要它）；
3. Doris Pipeline 连接器已内置 MySQL 协议依赖，无需额外驱动；SQL API 方式写 Doris 时按需添加 JDBC 驱动。

3.4.3 部署清单汇总

/opt/apps/ ├── flink/ # Flink 1.20.1 │ ├── bin/{start-cluster.sh, sql-client.sh, flink} │ ├── conf/config.yaml │ └── lib/ # SQL API 依赖（见 3.4.2） ├── flink-cdc-3.5.0/ # Flink CDC 发行包 │ ├── bin/flink-cdc.sh │ ├── lib/ # Pipeline 连接器（内置） │ └── conf/ # Pipeline YAML ├── mysql-8.0.x / 或外部 RDS # 数据源 ├── doris-4.1/ (FE + BE) # OLAP Sink └── kafka-3.7/ # 消息 Sink

3.5 课堂环境快速拉起（Docker Compose）

为了让大家零依赖、10 分钟内搭好实验环境，本节提供一键 Docker Compose 套件（MySQL + Doris + Kafka）。Paimon 在开发模式直接使用本地文件系统，无需额外部署。

3.5.1 docker-compose.yml

version:"3.8"services:# ===== MySQL 数据源（Binlog 已开启）=====mysql:image:mysql:8.0container_name:cdc-mysqlports:-"3306:3306"environment:MYSQL_ROOT_PASSWORD:"Root.2026"MYSQL_DATABASE:appcommand:--server-id=1--log-bin=mysql-bin--binlog-format=ROW--binlog-row-image=FULL--gtid-mode=ON--enforce-gtid-consistency=ON--default-authentication-plugin=mysql_native_passwordvolumes:-mysql-data:/var/lib/mysqlhealthcheck:test:["CMD","mysqladmin","ping","-uroot","-pRoot.2026"]interval:10sretries:5# ===== Doris 4.1 FE =====doris-fe:image:apache/doris:fe-4.1.0-x86_64container_name:cdc-doris-feports:-"8030:8030"# Stream Load-"9030:9030"# MySQL 协议查询environment:FE_SERVERS:"fe1:cdc-doris-fe:9010"FE_ID:1volumes:-doris-fe-data:/opt/apache-doris/fe/doris-meta# ===== Doris 4.1 BE =====doris-be:image:apache/doris:be-4.1.0-x86_64container_name:cdc-doris-beports:-"8040:8040"# BE HTTPenvironment:FE_SERVERS:"fe1:cdc-doris-fe:9010"BE_ADDR:"cdc-doris-be:9050"volumes:-doris-be-data:/opt/apache-doris/be/storagedepends_on:-doris-fe# ===== Kafka（KRaft 模式，免 Zookeeper）=====kafka:image:bitnami/kafka:3.7container_name:cdc-kafkaports:-"9092:9092"environment:KAFKA_CFG_NODE_ID:1KAFKA_CFG_PROCESS_ROLES:broker,controllerKAFKA_CFG_CONTROLLER_QUORUM_VOTERS:1@kafka:9093KAFKA_CFG_LISTENERS:PLAINTEXT://:9092,CONTROLLER://:9093KAFKA_CFG_ADVERTISED_LISTENERS:PLAINTEXT://kafka:9092KAFKA_CFG_CONTROLLER_LISTENER_NAMES:CONTROLLERKAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE:"true"volumes:-kafka-data:/bitnami/kafkavolumes:mysql-data:doris-fe-data:doris-be-data:kafka-data:

注意：

1. Doris 镜像 tag 以 Docker Hub 官方仓库 实际发布的 4.1 tag 为准；
2. Doris BE 容器内存建议 ≥ 4G，宿主机内存不足时调低BE的mem_limit；
3. 生产环境请勿使用容器内单节点 Doris，按集群规划部署。

3.5.2 一键启动与初始化

# 1. 启动全部服务dockercompose up-d# 2. 等待健康检查通过（约 1-2 分钟）dockercomposeps# 3. 初始化 MySQL 业务数据dockerexec-icdc-mysql mysql-uroot-pRoot.2026<<'EOF' CREATE USER 'cdc_user'@'%' IDENTIFIED BY 'Cdc.2026'; GRANT SELECT, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'cdc_user'@'%'; FLUSH PRIVILEGES; CREATE DATABASE app DEFAULT CHARACTER SET utf8mb4; USE app; CREATE TABLE orders ( id BIGINT NOT NULL AUTO_INCREMENT, customer_id BIGINT NOT NULL, amount DECIMAL(10,2) NOT NULL, status VARCHAR(20) NOT NULL DEFAULT 'created', create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id) ) ENGINE = InnoDB; INSERT INTO orders (customer_id, amount, status) VALUES (1001, 99.90, 'paid'), (1002, 199.00, 'created'), (1003, 59.50, 'paid'); EOF# 4. 注册 Doris BE 到 FEdockerexeccdc-doris-be mysql-hcdc-doris-fe-P9030-uroot\-e"ALTER SYSTEM ADD BACKEND 'cdc-doris-be:9050';"

3.5.3 各组件快速验证

# MySQL：验证 Binlog 与权限dockerexec-icdc-mysql mysql-ucdc_user-pCdc.2026\-e"SHOW VARIABLES LIKE 'log_bin'; SHOW GRANTS;"# Doris：验证 FE/BE 状态dockerexeccdc-doris-be mysql-hcdc-doris-fe-P9030-uroot\-e"SHOW FRONTENDS\G SHOW BACKENDS\G"# Kafka：验证 Brokerdockerexeccdc-kafka kafka-topics.sh --bootstrap-server kafka:9092--list

3.6 环境连通性验证

在进入实战章节前，用下面的Checklist逐项确认环境就绪。任何一项不通，先解决再继续，否则后续章节会连环报错。

3.6.1 连通性验证清单

3.6.2 常见环境问题速查

3.7 本章小结

本章完成了整条实战链路的环境搭建：

1. 版本矩阵：锁定 Flink CDC 3.5.0 + Flink 1.20 + MySQL 8.0/8.4 +Doris 4.1+ Paimon 1.4.2+ + Kafka 3.x 的基线组合，牢记连接器兼容性是第一踩坑点。
2. MySQL 配置：Binlog（ROW + FULL）、GTID、专用账号四项复制权限，业务表必须有主键——这是 CDC 数据源的三板斧。
3. Flink 集群：Standalone 为主线，config.yaml核心配置（Checkpoint + 重启策略），并预告 YARN / K8s 生产化方式。
4. 依赖部署：Flink CDC 发行包内置四大 Pipeline 连接器（flink-cdc.sh提交即用）；SQL API 方式则按需向 Flinklib/添加连接器 jar。
5. 快速环境：Docker Compose 一键拉起 MySQL + Doris 4.1 + Kafka，Paimon 开发态用本地文件系统。
6. 连通性验证：12 项 Checklist + 高频问题速查，环境通则后续顺。

下一章预告：第 4 章《MySQL CDC Source 全方位解析》将以本章环境为基础，逐参数讲解 MySQL Pipeline Source 的配置、启动模式与快照调优，并用 Print Sink 完成第一次端到端验证。

参考资料

• Flink CDC 3.5.0 官方文档：https://nightlies.apache.org/flink/flink-cdc-docs-release-3.5/
• Flink 1.20 下载页：https://flink.apache.org/downloads/
• Flink 1.20 配置文档：https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/deployment/config/
• Doris 官网（4.1）：https://doris.apache.org/
• Doris Docker 部署：https://doris.apache.org/docs/install/construct-docker/
• Paimon 下载页：https://paimon.apache.org/docs/master/project/download/
• Kafka 下载页：https://kafka.apache.org/downloads

本文是《Flink CDC 3.5.0 从理论到实践》系列教程的第 3 章，后续章节将持续更新，欢迎关注收藏。