完整教程：从 “T+1” 到 “秒级”：MySQL+Flink+Doris 构建实时数据分析全链路

一、背景：传统数据处理的痛点催生新组合

随着业务数字化深入，企业对 “数据价值挖掘” 的需求从 “离线统计” 转向 “实时洞察”，但传统数据处理方案存在明显短板，成为业务增长的瓶颈：

1. OLTP 与 OLAP 资源争抢：传统方案中，业务交易（如 MySQL 下单）与分析查询（如统计月度销量）共用一套数据库，复杂分析 SQL 会占用大量 CPU/IO，导致线上订单响应变慢（如用户下单卡顿）；
2. 数据同步延迟高：依赖 “定时 ETL 脚本（如 Shell/DataX）” 从 MySQL 拉取数据，同步周期通常是 “小时级” 甚至 “T+1”，无法满足实时需求（如制造业需实时监控订单生产进度、电商需实时更新销量看板）；
3. 业务侵入性强：增量同步时需通过 “查询 MySQL 表（如按 update_time 过滤）” 实现，高峰时段会加重业务库负担，甚至引发锁表；
4. 数据一致性难保障：网络中断、脚本重试等场景易导致数据重复 / 丢失（如订单统计多算 / 少算），影响分析结果可信度；
5. 扩展性差：单机 MySQL 无法支撑海量历史数据，离线分析库（如 Hive）查询延迟高，难以应对 “日订单量从千级到万级” 的业务增长。

正是这些痛点，推动了 “MySQL（业务）+Flink-CDC（抓变更）+Flink（处理）+Doris（分析）” 这套实时数据链路的普及 —— 它本质是为解决 “业务不中断、数据实时用、分析高效率” 的核心矛盾而生。

二、作用：构建 “数据产生→同步→处理→分析” 全链路闭环

这套组合的核心作用是打通 “业务交易数据” 到 “实时分析” 的链路，每个组件各司其职，形成无断点的数据流闭环，具体作用可分为 “组件分工” 和 “整体价值”：

2.1. 组件分工：每个环节精准定位

组件	核心作用	具体动作
MySQL 8	业务交易底座（OLTP）	存储线上核心业务数据（如机械订单、用户支付记录），支撑下单、改单、支付等实时操作，保证交易低延迟；
Flink-CDC 3.2.1	无侵入式数据变更捕获	读取 MySQL 的 binlog 日志（MySQL 8 默认支持 ROW 格式），实时捕获数据新增 / 修改 / 删除，不发起业务表查询，无侵入；
Flink 1.20.2	实时数据处理中枢	① 接收 Flink-CDC 的变更数据，做清洗（去重、补全缺失字段）、转换（格式对齐、关联维度表）；② 保证数据同步一致性（Exactly-Once 语义，重启不丢不重）；③ 按配置将数据实时写入 Doris；
Doris 2.1.10	高效 OLAP 分析引擎	① 存储同步来的业务数据，支持高压缩比（降低存储成本）；② 提供毫秒 / 秒级复杂查询（如 “机械订单按区域 + 月份聚合销量”）；③ 支持报表、Dashboard、adhoc 查询，满足多维度分析需求；

2.2. 整体价值：从 “数据静止” 到 “数据流动”

传统方案中，数据是 “静止的”（MySQL 里的订单数据，要等定时脚本拉取后才能分析）；而这套组合让数据 “流动起来”——MySQL 产生新订单→Flink-CDC 毫秒级捕获→Flink 处理后秒级写入 Doris→Doris 即时支持分析，实现 “数据产生即能用”。

三、需求：哪些场景必须用这套组合？

1. 实时决策需求：业务需秒级获取数据反馈（如电商实时销售大屏、金融实时风控），替代传统小时级延迟的批量数据架构；
2. 数据一致性需求：数据同步、处理过程需 “不丢不重”，避免因数据错误导致决策偏差（如订单统计漏算）；
3. 高效分析需求：需支撑百亿级数据的高并发查询（如多分析师同时查实时报表），且结果秒级返回；
4. 低门槛使用需求：减少技术学习成本，分析师可通过熟悉的 SQL（兼容 MySQL 协议）直接做分析，无需适配新工具。

典型适用场景

• 电商实时运营：实时销售大屏、商品库存预警、用户实时画像；
• 金融实时风控：实时监控交易异常（如同一账户异地高频转账）、信贷风险指标计算；
• 零售实时库存：实时跟踪各门店库存变化，触发补货提醒；
• 互联网实时用户运营：实时计算用户活跃度、APP 崩溃率，及时调整运营策略。

四、核心优势（技术栈亮点）

1. 实时性强：Flink-CDC 秒级捕获 MySQL 变更，Flink 实时处理，Doris 实时查询，全链路延迟控制在秒级；
2. 数据可靠：支持 Exactly-Once 语义，MySQL 事务 + Flink 容错 + Doris 原子写入，保证数据不丢不重；
3. 分析高效：Doris 基于 MPP 架构，百亿级数据并行计算，即席查询 / 报表分析秒级出结果；
4. 易用性高：端到端兼容 MySQL 生态（MySQL 存数据、Doris 用 MySQL 协议查数据），降低开发与分析门槛；
5. 扩展性好：各组件均支持水平扩容（Flink 加 TaskManager、Doris 加 BE 节点），轻松应对数据量增长。

五、构建实时数据驱动架构

本实验以测试环境为主，一些服务的配置大小不符合生产环境的标准，还请在使用时注意修改自己的配置！

主机名	IP地址	配置
MySQL	10.0.0.6/24	4G/4C/100GB
Doris	10.0.0.7/24	4G/2C/100GB
Flink	10.0.0.8/24	4G/2C/100GB

操作系统为Ubuntu22.04.4 LTS，防火墙已经关闭。

5.1 二进制部署MySQL

5.1.1 上传MySQL8.0.25二进制包

[root@mysql ~]# ls
mysql-8.0.25-linux-glibc2.12-x86_64.tar.xz
# 解压二进制包到/usr/local/
[root@mysql ~]# tar -xf mysql-8.0.25-linux-glibc2.12-x86_64.tar.xz -C /usr/local/
# 进入/usr/local/目录，给MySQL做一个软链接
[root@mysql ~]# cd /usr/local/
[root@mysql /usr/local]# ln -s mysql-8.0.25-linux-glibc2.12-x86_64/ mysql
[root@mysql /usr/local]# ll
total 44
lrwxrwxrwx  1 root root   36 Sep 22 02:48 mysql -> mysql-8.0.25-linux-glibc2.12-x86_64//
drwxr-xr-x  9 root root 4096 Sep 22 02:48 mysql-8.0.25-linux-glibc2.12-x86_64/
# 创建MySQL的管理用户
[root@mysql ~]# useradd mysql -M -s /sbin/nologin
# 创建数据库服务数据目录
[root@mysql ~]# mkdir -p /data/mysql/{data,log}
[root@mysql ~]# chown mysql.mysql /data/mysql/data
[root@mysql ~]# chown mysql.mysql /data/mysql/log
[root@mysql /etc]# ll /data/mysql/
total 16
drwxr-xr-x 4 root  root  4096 Sep 22 02:57 ./
drwxr-xr-x 3 root  root  4096 Sep 22 02:52 ../
drwxr-xr-x 2 mysql mysql 4096 Sep 22 02:52 data/
drwxr-xr-x 2 mysql mysql 4096 Sep 22 02:57 log/
# 创建MySQL环境变量
[root@mysql ~]# ln -s /usr/lib/x86_64-linux-gnu/libtinfo.so.6 /usr/lib/x86_64-linux-gnu/libtinfo.so.5
[root@mysql ~]# tail -1 /etc/profile
export PATH="$PATH:/usr/local/mysql/bin"
[root@mysql ~]# mysql -V
mysql  Ver 8.0.25 for Linux on x86_64 (MySQL Community Server - GPL)

5.1.2 初始化MySQL

在写之前，我个人习惯先将my.cnf配置好，初始化的时候按照my.cnf配置去初始化,配置大小可以按照自己的机器配置调整。

[root@mysql ~]# cat /etc/my.cnf
[mysqld]
# ===================== 基础环境配置 =====================
server-id=100
# MySQL 安装目录（你的软链接路径）
basedir = /usr/local/mysql
# 数据存储目录（已存在，权限为 mysql:mysql）
datadir = /data/mysql/data
# 数据库服务端口（默认3306，如需修改需同步防火墙）
port = 3306
# 运行用户（与数据目录权限一致，避免权限报错）
user = mysql
# 套接字文件路径（客户端连接用，需与客户端配置匹配）
socket = /tmp/mysql.sock
# 服务ID（开启二进制日志必须配置，单实例设为1即可，主从需唯一）
lower_case_table_names = 1
sql-mode=STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_ENGINE_SUBSTITUTION
#sql_mode=STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION
# ===================== 内存优化配置 =====================
# InnoDB 缓冲池（核心参数，占总内存 50%-70%，15Gi 内存设为 8G，避免内存溢出）
innodb_buffer_pool_size = 2G
# 缓冲池分片（超过4G建议分片，每片不超过4G）
innodb_buffer_pool_instances = 1
# 临时表最大内存（避免临时表溢出到磁盘，与 max_heap_table_size 保持一致）
tmp_table_size = 64M
max_heap_table_size = 64M
# 连接缓存大小（根据连接数调整，1000连接设为 64M）
join_buffer_size = 2M
sort_buffer_size = 2M
read_buffer_size = 2M
read_rnd_buffer_size = 4M
# ===================== 连接与并发配置 =====================
# 最大并发连接数（根据业务调整，15Gi内存支持 1000-2000 连接，默认设1000）
max_connections = 100
# 最大等待连接数（超过 max_connections 时的排队数）
back_log = 20
# 空闲连接超时时间（避免空闲连接占用资源，设为1小时）
wait_timeout = 3600
interactive_timeout = 3600
# 禁止 DNS 解析（加快客户端连接，避免因 DNS 问题导致连接慢）
skip_name_resolve = 1
# ===================== InnoDB 核心配置 =====================
# 事务日志文件大小（建议 256M-1G，512M 兼顾性能与恢复速度）
innodb_log_file_size = 256M
# 事务日志组数量（默认2个，无需修改）
innodb_log_files_in_group = 2
# 事务日志存储路径（与数据目录一致，如需单独存放可修改）
innodb_log_group_home_dir = /data/mysql/data
# 事务刷盘策略（1=每次事务提交刷盘，最安全；0=每秒刷盘，性能高但可能丢数据）
innodb_flush_log_at_trx_commit = 1
# 独立表空间（每个表一个 .ibd 文件，方便单表备份/迁移，MySQL 8.0+ 默认开启，显式配置更安全）
innodb_file_per_table = 1
# 磁盘IO模式（O_DIRECT 跳过系统缓存，减少IO冗余，适合生产环境）
innodb_flush_method = O_DIRECT
# 并发读写控制（自动调整，无需手动设值）
innodb_read_io_threads = 8
innodb_write_io_threads = 8
# 死锁检测（开启后自动检测死锁并回滚，避免业务卡住）
innodb_deadlock_detect = ON
# ===================== 日志配置（关键！用于问题排查与恢复） =====================
# 错误日志（记录启动/运行错误，必须开启，路径建议与数据目录分离）
log_error = /data/mysql/log/mysql_error.log
# 慢查询日志（记录执行时间超过 2 秒的SQL，优化业务SQL用）
slow_query_log = ON
slow_query_log_file = /data/mysql/log/mysql_slow.log
long_query_time = 2
# 二进制日志（用于主从复制、数据恢复，必须开启）
log_bin = /data/mysql/log/mysql_binlog
# 二进制日志格式（row 格式：记录数据行变化，主从复制最安全，避免SQL兼容性问题）
binlog_format = row
# 二进制日志过期时间（7天自动清理，避免磁盘占满，根据备份策略调整）
binlog_expire_logs_seconds = 604800
# 中继日志（主从复制时从库用，默认即可）
relay_log = /data/mysql/log/mysql_relaylog
# ===================== 客户端配置（[mysql] 段，客户端连接时生效） =====================
[mysql]
default-character-set = utf8mb4
socket = /tmp/mysql.sock
# ===================== 安全模式配置（[mysqld_safe] 段，启动异常时用） =====================
[mysqld_safe]
log-error = /data/mysql/log/mysql_error.log
pid-file = /data/mysql/data/mysql.pid

然后进行初始化

# 初始化数据库
[root@mysql ~]# mysqld --defaults-file=/etc/my.cnf --initialize-insecure --user=mysql --datadir=/data/mysql/data --basedir=/usr/local/mysql
[root@mysql ~]# ll /data/mysql/data/
total 602952
drwxr-xr-x 6 mysql mysql      4096 Sep 22 02:59  ./
drwxr-xr-x 4 root  root       4096 Sep 22 02:57  ../
-rw-r----- 1 mysql mysql        56 Sep 22 02:59  auto.cnf
-rw------- 1 mysql mysql      1676 Sep 22 02:59  ca-key.pem
-rw-r--r-- 1 mysql mysql      1112 Sep 22 02:59  ca.pem
-rw-r--r-- 1 mysql mysql      1112 Sep 22 02:59  client-cert.pem
-rw------- 1 mysql mysql      1676 Sep 22 02:59  client-key.pem
-rw-r----- 1 mysql mysql    393216 Sep 22 02:59 '#ib_16384_0.dblwr'
-rw-r----- 1 mysql mysql   8781824 Sep 22 02:59 '#ib_16384_1.dblwr'
-rw-r----- 1 mysql mysql      5931 Sep 22 02:59  ib_buffer_pool
-rw-r----- 1 mysql mysql  12582912 Sep 22 02:59  ibdata1
-rw-r----- 1 mysql mysql 268435456 Sep 22 02:59  ib_logfile0
-rw-r----- 1 mysql mysql 268435456 Sep 22 02:59  ib_logfile1
drwxr-x--- 2 mysql mysql      4096 Sep 22 02:59 '#innodb_temp'/
drwxr-x--- 2 mysql mysql      4096 Sep 22 02:59  mysql/
-rw-r----- 1 mysql mysql  25165824 Sep 22 02:59  mysql.ibd
drwxr-x--- 2 mysql mysql      4096 Sep 22 02:59  performance_schema/
-rw------- 1 mysql mysql      1676 Sep 22 02:59  private_key.pem
-rw-r--r-- 1 mysql mysql       452 Sep 22 02:59  public_key.pem
-rw-r--r-- 1 mysql mysql      1112 Sep 22 02:59  server-cert.pem
-rw------- 1 mysql mysql      1676 Sep 22 02:59  server-key.pem
drwxr-x--- 2 mysql mysql      4096 Sep 22 02:59  sys/
-rw-r----- 1 mysql mysql  16777216 Sep 22 02:59  undo_001
-rw-r----- 1 mysql mysql  16777216 Sep 22 02:59  undo_002
[root@mysql ~]# ll /data/mysql/log/
total 24
drwxr-xr-x 2 mysql mysql 4096 Sep 22 02:59 ./
drwxr-xr-x 4 root  root  4096 Sep 22 02:57 ../
-rw-r----- 1 mysql mysql  179 Sep 22 02:59 mysql_binlog.000001
-rw-r----- 1 mysql mysql   36 Sep 22 02:59 mysql_binlog.index
-rw-r----- 1 mysql mysql  549 Sep 22 02:59 mysql_error.log
-rw-r----- 1 mysql mysql  207 Sep 22 02:59 mysql_slow.log

5.1.3 启动运行数据库服务

# 利用shell编写脚本文件启动
[root@mysql ~]# cp /usr/local/mysql/support-files/mysql.server /etc/init.d/mysqld
[root@mysql ~]# file /usr/local/mysql/support-files/mysql.server
/usr/local/mysql/support-files/mysql.server: POSIX shell script, ASCII text executable
# 使用systemd管理进程
[root@mysql ~]# cat /usr/lib/systemd/system/mysqld.service
# /usr/lib/systemd/system/mysqld.service
[Unit]
Description=MySQL Server Daemon
Documentation=https://dev.mysql.com/doc/
After=network.target remote-fs.target nss-lookup.target
[Service]
Type=forking
ExecStart=/etc/init.d/mysqld start
ExecStop=/etc/init.d/mysqld stop
Restart=on-failure
RestartSec=3
LimitNOFILE=65535
[Install]
WantedBy=multi-user.target
# 启动数据库
[root@mysql ~]# systemctl daemon-reload
[root@mysql ~]# systemctl enable mysqld
Synchronizing state of mysqld.service with SysV service script with /lib/systemd/systemd-sysv-install.
Executing: /lib/systemd/systemd-sysv-install enable mysqld
Created symlink /etc/systemd/system/multi-user.target.wants/mysqld.service → /lib/systemd/system/mysqld.service.
[root@mysql ~]# systemctl start mysqld
[root@mysql ~]# systemctl status mysqld
● mysqld.service - MySQL Server DaemonLoaded: loaded (/lib/systemd/system/mysqld.service; enabled; vendor preset: enabled)Active: active (running) since Mon 2025-09-22 03:17:49 UTC; 3s agoDocs: https://dev.mysql.com/doc/Process: 26733 ExecStart=/etc/init.d/mysqld start (code=exited, status=0/SUCCESS)Main PID: 26766 (mysqld_safe)Tasks: 47 (limit: 4515)Memory: 511.4MCPU: 1.454sCGroup: /system.slice/mysqld.service├─26766 /bin/sh /usr/local/mysql/bin/mysqld_safe --datadir=/data/mysql/data --pid-file=/data/mysql/data/mysql.pid└─27367 /usr/local/mysql/bin/mysqld --basedir=/usr/local/mysql --datadir=/data/mysql/data --plugin-dir=/usr/local/mysql/lib/plugin --user=mysql>
Sep 22 03:17:47 mysql systemd[1]: Starting MySQL Server Daemon...
Sep 22 03:17:47 mysql mysqld[26733]: Starting MySQL
Sep 22 03:17:49 mysql mysqld[26733]: .. *
Sep 22 03:17:49 mysql systemd[1]: Started MySQL Server Daemon.

5.1.4.新建数据库密码

这里因为是测试环境，所以就简单一点了。

[root@mysql ~]# mysqladmin -uroot password '123456'
mysqladmin: [Warning] Using a password on the command line interface can be insecure.
Warning: Since password will be sent to server in plain text, use ssl connection to ensure password safety.
[root@mysql ~]# mysql -uroot -p123456
mysql: [Warning] Using a password on the command line interface can be insecure.
Welcome to the MySQL monitor.  Commands end with ; or \g.
Your MySQL connection id is 9
Server version: 8.0.25 MySQL Community Server - GPL
Copyright (c) 2000, 2021, Oracle and/or its affiliates.
Oracle is a registered trademark of Oracle Corporation and/or its
affiliates. Other names may be trademarks of their respective
owners.
Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.
mysql> exit
Bye

到此数据库就安装成功了！

六、二进制部署Doris

6.1 安装java11环境&&关闭swap&&调优内核参数

[root@doris ~]# cat /etc/apt/sources.list
deb https://mirrors.aliyun.com/ubuntu/ jammy main restricted universe multiverse
deb-src https://mirrors.aliyun.com/ubuntu/ jammy main restricted universe multiverse
deb https://mirrors.aliyun.com/ubuntu/ jammy-security main restricted universe multiverse
deb-src https://mirrors.aliyun.com/ubuntu/ jammy-security main restricted universe multiverse
deb https://mirrors.aliyun.com/ubuntu/ jammy-updates main restricted universe multiverse
deb-src https://mirrors.aliyun.com/ubuntu/ jammy-updates main restricted universe multiverse
# deb https://mirrors.aliyun.com/ubuntu/ jammy-proposed main restricted universe multiverse
# deb-src https://mirrors.aliyun.com/ubuntu/ jammy-proposed main restricted universe multiverse
deb https://mirrors.aliyun.com/ubuntu/ jammy-backports main restricted universe multiverse
deb-src https://mirrors.aliyun.com/ubuntu/ jammy-backports main restricted universe multiverse
[root@doris ~]# apt update
[root@doris ~]# apt install openjdk-11-jdk -y
# 由此安装完成之后JAVA_HOME就在
/usr/lib/jvm/java-11-openjdk-amd64/
# 关闭swap
[root@doris ~]# swapoff -a
[root@doris ~]# sed -i '/swap/s/^/#/' /etc/fstab
# 设置ulimit值
[root@doris /usr/local/doris/be]# vim /etc/security/limits.conf
* soft nofile 655350
* hard nofile 655350
root soft nofile 655350
root hard nofile 655350
[root@doris /usr/local/doris/be]# systemctl daemon-reexec
# 关闭THP透明大页
[root@doris /usr/local/doris/be]# echo never > /sys/kernel/mm/transparent_hugepage/enabled
[root@doris /usr/local/doris/be]# echo never > /sys/kernel/mm/transparent_hugepage/defrag
# 设置vm.max_map_count
# 编辑sysctl配置文件
vim /etc/sysctl.conf
# 添加以下行
vm.max_map_count=2000000
# 使配置生效
sysctl -p
### 验证
# 验证ulimit
ulimit -n
# 验证透明大页状态
cat /sys/kernel/mm/transparent_hugepage/enabled
# 验证swap状态
free -h
# 验证vm.max_map_count
sysctl vm.max_map_count

6.2 二进制部署Doris2.1.10

[root@doris ~]# ls
apache-doris-2.1.10-bin-x64.tar.gz
# 解压到/usr/loca/目录下，并且做上软链接
[root@doris ~]# tar xf apache-doris-2.1.10-bin-x64.tar.gz -C /usr/local/
[root@doris ~]# ln -s /usr/local/apache-doris-2.1.10-bin-x64/ /usr/local/doris
[root@doris ~]# ll /usr/local/
total 44
drwxr-xr-x 11 root root 4096 Sep 22 03:36 ./
drwxr-xr-x 14 root root 4096 Feb 16  2024 ../
drwxr-xr-x  5 root root 4096 May 14 14:24 apache-doris-2.1.10-bin-x64/
drwxr-xr-x  2 root root 4096 Feb 16  2024 bin/
lrwxrwxrwx  1 root root   39 Sep 22 03:36 doris -> /usr/local/apache-doris-2.1.10-bin-x64/
# 创建doris数据目录
[root@doris /usr/local/doris/fe]# mkdir -p /data/{storage,doris-meta}
[root@doris /usr/local/doris/fe]# ll /data/
total 16
drwxr-xr-x  4 root root 4096 Sep 22 03:38 ./
drwxr-xr-x 21 root root 4096 Sep 22 03:38 ../
drwxr-xr-x  2 root root 4096 Sep 22 03:38 doris-meta/
drwxr-xr-x  2 root root 4096 Sep 22 03:38 storage/

6.2.1 启动FE

# 修改doris的FE和BE的配置
主要两处
[root@doris ~]# vim /usr/local/doris/fe/conf/fe.conf
meta_dir = /data/doris-meta
...
priority_networks = 10.0.0.0/24
# 解释参数含义
meta_dir = /data/doris-meta
作用：指定 FE 存储元数据的目录。
priority_networks = 10.0.0.0/24
作用：指定 FE 节点优先使用的网络地址范围，用于集群内部节点间通信（如 FE 与 BE、FE 与其他 FE 之间的交互）。
[root@doris ~]# cd /usr/local/doris/fe/
[root@doris /usr/local/doris/fe]# ./bin/start_fe.sh --daemon
[root@doris /usr/local/doris/fe]# jps
6232 Jps
6012 DorisFE
# 查看 FE 运行状态
[root@doris /usr/local/doris/fe]# curl http://127.0.0.1:8030/api/bootstrap;echo
{"msg":"success","code":0,"data":{"replayedJournalId":0,"queryPort":0,"rpcPort":0,"arrowFlightSqlPort":0,"version":""},"count":0}
# 这里 IP 和 端口分别是 FE 的 IP 和 http_port（默认8030），如果是你在 FE 节点执行，直接运行上面的命令即可。
# 如果返回结果中带有 "msg":"success" 字样，则说明启动成功。

6.2.2 连接FE

[root@mysql ~]# mysql -uroot -P9030 -h10.0.0.7
Welcome to the MySQL monitor.  Commands end with ; or \g.
Your MySQL connection id is 0
Server version: 5.7.99
Copyright (c) 2000, 2021, Oracle and/or its affiliates.
Oracle is a registered trademark of Oracle Corporation and/or its
affiliates. Other names may be trademarks of their respective
owners.
Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.
mysql>
# 执行下面的命令查看 FE 运行状态
mysql> show frontends\G;
*************************** 1. row ***************************Name: fe_1efea265_a533_446f_969c_3352d86802d8Host: 10.0.0.7EditLogPort: 9010HttpPort: 8030QueryPort: 9030RpcPort: 9020
ArrowFlightSqlPort: -1Role: FOLLOWERIsMaster: trueClusterId: 1666407969Join: trueAlive: trueReplayedJournalId: 69LastStartTime: 2025-09-22 03:40:48LastHeartbeat: 2025-09-22 03:45:48IsHelper: trueErrMsg:Version: doris-2.1.10-rc01-33df5ba180CurrentConnected: Yes
1 row in set (0.07 sec)
ERROR:
No query specified
# 如果 IsMaster、Join 和 Alive 三列均为true，则表示节点正常

6.2.3 配置并启动BE

[root@doris /usr/local/doris/be]# pwd
/usr/local/doris/be
# 修改BE配置如下三处
[root@doris /usr/local/doris/be]# vim conf/be.conf
JAVA_HOME=/usr/lib/jvm/java-11-openjdk-amd64/
priority_networks = 10.0.0.0/24
storage_root_path = /data/storage
# 启动BE
[root@doris /usr/local/doris/be]# ./bin/start_be.sh --daemon
[root@doris /usr/local/doris/be]# jps
12803 Jps
12777 DorisBE
6012 DorisFE
# 添加BE节点到集群
mysql> ALTER SYSTEM ADD BACKEND "10.0.0.7:9050";
Query OK, 0 rows affected (0.05 sec)
mysql> SHOW BACKENDS\G
*************************** 1. row ***************************BackendId: 11718Host: 10.0.0.7HeartbeatPort: 9050BePort: 9060HttpPort: 8040BrpcPort: 8060ArrowFlightSqlPort: -1LastStartTime: 2025-09-22 06:03:27LastHeartbeat: 2025-09-22 06:04:01Alive: trueSystemDecommissioned: falseTabletNum: 0DataUsedCapacity: 0.000TrashUsedCapacity: 0.000AvailCapacity: 29.058 GBTotalCapacity: 47.926 GBUsedPct: 39.37 %MaxDiskUsedPct: 39.37 %RemoteUsedCapacity: 0.000Tag: {"location" : "default"}ErrMsg:Version: doris-2.1.10-rc01-33df5ba180Status: {"lastSuccessReportTabletsTime":"N/A","lastStreamLoadTime":-1,"isQueryDisabled":false,"isLoadDisabled":false}
HeartbeatFailureCounter: 0NodeRole: mix
1 row in set (0.01 sec)
# Alive : true表示节点运行正常

6.2.4 访问测试

用户名root

默认没设置密码，因为这里是测试环境，我就不再设置了。

# 如果想设置密码可以参考如下
确认当前用户为 root，在 MySQL Client 中查看当前用户
mysql> select user();
+-------------------+
| user()            |
+-------------------+
| 'root'@'10.0.0.6' |
+-------------------+
1 row in set (0.04 sec)
# 修改 root 用户密码，在 MySQL Client 中执行 set password 命令
mysql> SET PASSWORD = PASSWORD('123456');
Query OK, 0 rows affected (0.02 sec)

Doris节点IP+8030端口

七、部署Flink+Flink-CDC

7.1 二进制部署flink1.20.2

7.1.1 上传二进制包

[root@flink ~]# ll
total 506652
drwx------  5 root root      4096 Sep 22 03:27 ./
drwxr-xr-x 20 root root      4096 Jun 12 09:59 ../
-rw-------  1 root root       135 Sep 22 02:22 .bash_history
-rw-r--r--  1 root root      3106 Oct 15  2021 .bashrc
drwx------  2 root root      4096 Jun 13 03:19 .cache/
-rw-r--r--  1 root root 484152134 Sep 13 08:59 flink-1.20.2-bin-scala_2.12.tgz
-rw-r--r--  1 root root  34608108 Sep 17 09:27 flink-cdc-3.2.1-bin.tar.gz
# 解压并且做软链接
[root@flink ~]# tar xf flink-1.20.2-bin-scala_2.12.tgz -C /usr/local/
[root@flink ~]# tar xf flink-cdc-3.2.1-bin.tar.gz -C /usr/local/
[root@flink ~]# ln -s /usr/local/flink-1.20.2/ /usr/local/flink
[root@flink ~]# ln -s /usr/local/flink-cdc-3.2.1/ /usr/local/flink-cdc
[root@flink ~]# ll /usr/local/
total 48
drwxr-xr-x 12 root root  4096 Sep 22 06:09 ./
drwxr-xr-x 14 root root  4096 Feb 16  2024 ../
drwxr-xr-x  2 root root  4096 Feb 16  2024 bin/
drwxr-xr-x  2 root root  4096 Feb 16  2024 etc/
lrwxrwxrwx  1 root root    24 Sep 22 06:09 flink -> /usr/local/flink-1.20.2//
drwxr-xr-x 10  501 staff 4096 Jun 13 11:05 flink-1.20.2/
lrwxrwxrwx  1 root root    27 Sep 22 06:09 flink-cdc -> /usr/local/flink-cdc-3.2.1//
drwxr-xr-x  6 root root  4096 Sep 22 06:09 flink-cdc-3.2.1/

7.1.2 配置flink

# 单节点部署配置参考
[root@flink /usr/local/flink/conf]# pwd
/usr/local/flink/conf
[root@flink /usr/local/flink/conf]# grep -Ev '^.*#|^$' config.yaml
env:java:opts:all: --add-exports=java.base/sun.net.util=ALL-UNNAMED --add-exports=java.rmi/sun.rmi.registry=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED --add-exports=java.security.jgss/sun.security.krb5=ALL-UNNAMED --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.text=ALL-UNNAMED --add-opens=java.base/java.time=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.locks=ALL-UNNAMED
jobmanager:bind-host: 0.0.0.0rpc:address: localhostport: 6123memory:process:size: 1000mexecution:failover-strategy: region
taskmanager:bind-host: 0.0.0.0host: localhostnumberOfTaskSlots: 2memory:process:size: 1024m
parallelism:default: 1
rest:address: localhostbind-address: 0.0.0.0

7.1.3 启动flink

[root@flink /usr/local/flink/conf]# cd ../bin/
[root@flink /usr/local/flink/bin]# ./start-cluster.sh
[root@flink /usr/local/flink/bin]# jps
7286 TaskManagerRunner
7399 Jps
6762 StandaloneSessionClusterEntrypoint

7.1.4 访问Flink测试

7.2 二进制部署flink-cdc3.2.1

7.2.1 配置flink-cdc

# MySQL数据库首先创建测试的数据库与同步用户
mysql> CREATE USER IF NOT EXISTS `flink`@`%` IDENTIFIED BY 'hG6kcQP76bh&U95';
Query OK, 0 rows affected (0.01 sec)
mysql> GRANT ALL PRIVILEGES ON *.* TO `flink`@`%`;
Query OK, 0 rows affected (0.00 sec)
mysql> FLUSH PRIVILEGES;
Query OK, 0 rows affected (0.00 sec)
mysql> CREATE DATABASE `doris_test` CHARACTER SET 'utf8mb4' COLLATE 'utf8mb4_0900_ai_ci';
Query OK, 1 row affected (0.01 sec)
# 编写同步文件
[root@flink /usr/local/flink-cdc/conf]# pwd
/usr/local/flink-cdc/conf
[root@flink /usr/local/flink-cdc/conf]# cat auto_sync_mysql_to_doris.yaml
source:type: mysqlhostname: 10.0.0.6port: 3306username: flinkpassword: 'hG6kcQP76bh&U95'server-id: 1010-1020tables: doris_test.\.*scan.newly-added-table.enabled: truejdbc.properties.allowPublicKeyRetrieval: true
sink:type: dorisfenodes: 10.0.0.7:8030username: rootpassword: ''table.create.properties.light_schema_change: truetable.create.properties.replication_num: 1
pipeline:name: 同步MySQL中doris_test库数据parallelism: 1schema.change.behavior: EVOLVE
execution:parallelism: 1checkpointing:externalized-checkpoint-retention: RETAIN_ON_CANCELLATIONcheckpoint-storage: filesysteminterval: 5mintimeout: 10minmode: EXACTLY_ONCEmin-pause-between-checkpoints: 2min

7.2.2 装置flink-cdc插件连接器

[root@flink /usr/local/flink-cdc/lib]# ll
total 131388
drwxr-xr-x 2 root root      4096 Sep 22 06:30 ./
drwxr-xr-x 6 root root      4096 Sep 22 06:09 ../
-rw-r--r-- 1 root root  39643320 Mar 27 06:10 flink-cdc-dist-3.2.0.jar
-rw-r--r-- 1  501 staff 39644076 Nov 21  2024 flink-cdc-dist-3.2.1.jar
-rwxr-xr-x 1 root root  12322656 Mar 27 06:10 flink-cdc-pipeline-connector-doris-3.2.0.jar*
-rwxr-xr-x 1 root root  21345935 Sep 19 07:52 flink-cdc-pipeline-connector-mysql-3.2.0.jar*
-rw-r--r-- 1 root root  21566086 Mar 27 06:10 flink-sql-connector-mysql-cdc-3.2.0.jar
-rw-r--r-- 1 root root  2475087 Sep 22 06:59 mysql-connector-java-8.0.27.jar

7.2.3 装置flink插件连接器

主要添加如下一个

[root@flink /usr/local/flink/lib]# ll -t
total 205928
drwxr-xr-x  2  501 staff      4096 Sep 22 06:32 ./
-rw-r--r--  1 root root    2475087 Sep 22 06:32 mysql-connector-java-8.0.27.jar

7.2.4 MySQL数据库创建测试数据

mysql> USE doris_test;
mysql> CREATE TABLE `order_info` (`order_id` bigint(20) NOT NULL COMMENT '订单ID，唯一标识',`user_id` bigint(20) NOT NULL COMMENT '用户ID，关联用户表',`order_status` tinyint(4) NOT NULL COMMENT '订单状态：0-待付款，1-已付款，2-已发货，3-已签收，4-已取消，5-退款中，6-已退款',`total_amount` decimal(10,2) NOT NULL COMMENT '订单总金额（含运费）',`pay_amount` decimal(10,2) NOT NULL COMMENT '实际支付金额',`freight_amount` decimal(10,2) NOT NULL DEFAULT 0.00 COMMENT '运费金额',`discount_amount` decimal(10,2) NOT NULL DEFAULT 0.00 COMMENT '折扣金额',`pay_type` tinyint(4) DEFAULT NULL COMMENT '支付方式：1-支付宝，2-微信，3-银行卡，4-货到付款',`payment_time` datetime DEFAULT NULL COMMENT '支付时间',`delivery_time` datetime DEFAULT NULL COMMENT '发货时间',`receive_time` datetime DEFAULT NULL COMMENT '确认收货时间',`comment_time` datetime DEFAULT NULL COMMENT '评价时间',`create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '订单创建时间',`update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '订单更新时间',`receiver_name` varchar(50) NOT NULL COMMENT '收货人姓名',`receiver_phone` varchar(20) NOT NULL COMMENT '收货人电话',`receiver_address` varchar(500) NOT NULL COMMENT '收货地址',`order_note` varchar(500) DEFAULT NULL COMMENT '订单备注',`tracking_no` varchar(100) DEFAULT NULL COMMENT '物流单号',PRIMARY KEY (`order_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='电商订单信息表';
mysql> INSERT INTO `order_info` (`order_id`, `user_id`, `order_status`, `total_amount`, `pay_amount`, `freight_amount`, `discount_amount`, `pay_type`, `payment_time`, `delivery_time`, `receive_time`, `create_time`, `receiver_name`, `receiver_phone`, `receiver_address`)-> VALUES-> (100001, 1001, 3, 299.00, 299.00, 10.00, 0.00, 1, '2023-09-20 14:30:00', '2023-09-20 16:45:00', '2023-09-22 09:15:00', '2023-09-20 14:25:30', '张三', '13800138000', '北京市朝阳区XX街道XX小区1号楼1单元101'),-> (100002, 1002, 1, 5999.00, 5999.00, 0.00, 100.00, 2, '2023-09-21 09:10:00', NULL, NULL, '2023-09-21 09:05:12', '李四', '13900139000', '上海市浦东新区XX路XX号');

7.2.5 flink-cdc启动同步

# 添加如下配置
[root@flink /usr/local/flink-cdc]# pwd
/usr/local/flink-cdc
[root@flink /usr/local/flink-cdc]# grep -i 'export FLINK_HOME' ./bin/flink-cdc.sh
export FLINK_HOME="/usr/local/flink"
# 启动同步
[root@flink /usr/local/flink-cdc]# ./bin/flink-cdc.sh conf/auto_sync_mysql_to_doris.yaml
Pipeline has been submitted to cluster.
Job ID: 211f14f0294192ba0e1c26ca238b7e0a
Job Description: 同步MySQL中doris_test库数据

7.2.6 flink查看同步状态

八、验证同步

8.1 查看doris

CREATE TABLE IF NOT EXISTS mechanical_order (-- 核心主键与业务标识order_id BIGINT NOT NULL COMMENT '机械订单ID（唯一主键）',order_no VARCHAR(50) NOT NULL COMMENT '订单业务编号（如：JX20250922001）',user_id BIGINT NOT NULL COMMENT '下单客户ID',-- 机械产品信息product_type VARCHAR(50) NOT NULL COMMENT '产品类型（如：数控机床、液压设备）',product_model VARCHAR(100) NOT NULL COMMENT '设备型号（如：CK6150数控车床）',specification TEXT NOT NULL COMMENT '技术参数（如：主轴转速3000r/min，加工直径500mm）',quantity INT NOT NULL COMMENT '订购数量',unit VARCHAR(20) NOT NULL COMMENT '单位（台/套/组）',-- 供应商信息supplier_id BIGINT COMMENT '供应商ID',supplier_name VARCHAR(100) NOT NULL COMMENT '供应商名称',supplier_region VARCHAR(50) COMMENT '供应商地区（如：上海市浦东新区）',-- 价格与支付unit_price DECIMAL(12,2) NOT NULL COMMENT '单价（元）',total_price DECIMAL(15,2) NOT NULL COMMENT '总价（元）',deposit_ratio DECIMAL(5,2) DEFAULT 0.3 COMMENT '预付款比例（默认30%）',deposit_amount DECIMAL(15,2) COMMENT '预付款金额',pay_type TINYINT DEFAULT 1 COMMENT '支付方式（1：电汇 2：信用证 3：承兑）',pay_status TINYINT DEFAULT 0 COMMENT '支付状态（0：未付款 1：部分付款 2：全款）',-- 生产与交付production_cycle INT COMMENT '生产周期（天）',plan_delivery_date DATE COMMENT '计划交付日期',actual_delivery_date DATE COMMENT '实际交付日期',delivery_method VARCHAR(50) COMMENT '交付方式（如：汽运、铁路、海运）',-- 安装与验收install_needed TINYINT DEFAULT 1 COMMENT '是否需要安装（1：是 0：否）',install_engineer VARCHAR(50) COMMENT '安装工程师',acceptance_result TINYINT COMMENT '验收结果（1：通过 2：未通过）',acceptance_date DATE COMMENT '验收日期',-- 订单状态与日志order_status TINYINT NOT NULL DEFAULT 0 COMMENT '订单状态（0：草稿 1：已确认 2：生产中 3：已交付 4：已验收 5：已取消）',creator VARCHAR(50) NOT NULL COMMENT '创建人',create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',updater VARCHAR(50) COMMENT '更新人',update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',remark TEXT COMMENT '备注（如：特殊定制要求、延期原因）',-- MySQL 主键与索引PRIMARY KEY (order_id),UNIQUE KEY uk_order_no (order_no),  -- 订单编号唯一KEY idx_user_id (user_id),         -- 按客户ID查询KEY idx_create_time (create_time)  -- 按创建时间查询
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='机械设备订单表';
# 插入一条数据
INSERT INTO mechanical_order (order_id, order_no, user_id,product_type, product_model, specification, quantity, unit,supplier_name, unit_price, total_price, deposit_ratio, deposit_amount,pay_type, pay_status, production_cycle, plan_delivery_date,order_status, creator, remark  -- 去掉 production_status
) VALUES (1001, 'JX20250922001', 10001,'数控机床', 'CK6150', '主轴转速3000r/min，加工直径500mm，功率15kW', 2, '台','上海精密机床厂', 185000.00, 370000.00, 0.3, 111000.00,1, 1, 45, '2025-11-05',1, '张三', '客户要求加装自动送料装置'  -- 去掉对应的值 1
);

8.2 使用navicat连接doris

8.3 再次写入数据测试同步

MySQL新增表

CREATE TABLE IF NOT EXISTS mechanical_order (-- 核心主键与业务标识order_id BIGINT NOT NULL COMMENT '机械订单ID（唯一主键）',order_no VARCHAR(50) NOT NULL COMMENT '订单业务编号（如：JX20250922001）',user_id BIGINT NOT NULL COMMENT '下单客户ID',-- 机械产品信息product_type VARCHAR(50) NOT NULL COMMENT '产品类型（如：数控机床、液压设备）',product_model VARCHAR(100) NOT NULL COMMENT '设备型号（如：CK6150数控车床）',specification TEXT NOT NULL COMMENT '技术参数（如：主轴转速3000r/min，加工直径500mm）',quantity INT NOT NULL COMMENT '订购数量',unit VARCHAR(20) NOT NULL COMMENT '单位（台/套/组）',-- 供应商信息supplier_id BIGINT COMMENT '供应商ID',supplier_name VARCHAR(100) NOT NULL COMMENT '供应商名称',supplier_region VARCHAR(50) COMMENT '供应商地区（如：上海市浦东新区）',-- 价格与支付unit_price DECIMAL(12,2) NOT NULL COMMENT '单价（元）',total_price DECIMAL(15,2) NOT NULL COMMENT '总价（元）',deposit_ratio DECIMAL(5,2) DEFAULT 0.3 COMMENT '预付款比例（默认30%）',deposit_amount DECIMAL(15,2) COMMENT '预付款金额',pay_type TINYINT DEFAULT 1 COMMENT '支付方式（1：电汇 2：信用证 3：承兑）',pay_status TINYINT DEFAULT 0 COMMENT '支付状态（0：未付款 1：部分付款 2：全款）',-- 生产与交付production_cycle INT COMMENT '生产周期（天）',plan_delivery_date DATE COMMENT '计划交付日期',actual_delivery_date DATE COMMENT '实际交付日期',delivery_method VARCHAR(50) COMMENT '交付方式（如：汽运、铁路、海运）',-- 安装与验收install_needed TINYINT DEFAULT 1 COMMENT '是否需要安装（1：是 0：否）',install_engineer VARCHAR(50) COMMENT '安装工程师',acceptance_result TINYINT COMMENT '验收结果（1：通过 2：未通过）',acceptance_date DATE COMMENT '验收日期',-- 订单状态与日志order_status TINYINT NOT NULL DEFAULT 0 COMMENT '订单状态（0：草稿 1：已确认 2：生产中 3：已交付 4：已验收 5：已取消）',creator VARCHAR(50) NOT NULL COMMENT '创建人',create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',updater VARCHAR(50) COMMENT '更新人',update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',remark TEXT COMMENT '备注（如：特殊定制要求、延期原因）',-- MySQL 主键与索引PRIMARY KEY (order_id),UNIQUE KEY uk_order_no (order_no),  -- 订单编号唯一KEY idx_user_id (user_id),         -- 按客户ID查询KEY idx_create_time (create_time)  -- 按创建时间查询
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='机械设备订单表';

插入一条数据

INSERT INTO mechanical_order (order_id, order_no, user_id,product_type, product_model, specification, quantity, unit,supplier_name, unit_price, total_price, deposit_ratio, deposit_amount,pay_type, pay_status, production_cycle, plan_delivery_date,order_status, creator, remark  -- 去掉 production_status
) VALUES (1001, 'JX20250922001', 10001,'数控机床', 'CK6150', '主轴转速3000r/min，加工直径500mm，功率15kW', 2, '台','上海精密机床厂', 185000.00, 370000.00, 0.3, 111000.00,1, 1, 45, '2025-11-05',1, '张三', '客户要求加装自动送料装置'  -- 去掉对应的值 1
);

查看flink

没有问题.

九、总结

                MySQL8+Doris2.1.10+Flink1.20.2+Flink-CDC3.2.1 技术栈，本质是一套为 “实时数据决策” 而生的企业级解决方案。它以 MySQL8 为业务数据源头，通过 Flink-CDC3.2.1 无侵入捕获数据变更，依托 Flink1.20.2 完成实时清洗、转换与聚合，最终将结构化数据存入 Doris2.1.10，支撑秒级分析与高并发查询，形成 “数据产生→捕获→处理→存储→分析” 的全链路实时闭环。

        这套架构的核心价值在于打破传统批量 ETL 的小时级延迟，以 “秒级响应” 满足电商实时大屏、金融风控、零售库存预警等场景需求；同时兼顾数据可靠性（全链路 Exactly-Once 语义）、易用性（兼容 MySQL 生态）与扩展性（各组件水平扩容），为企业从 “事后分析” 转向 “实时决策” 提供了轻量化且高效的技术支撑。

        普通 MySQL 读写分离若要支撑数据量增长，需依赖分库分表（如 Sharding-JDBC），但分库分表会增加架构复杂度（如跨库关联查询困难），且扩容时需重新拆分数据，成本高、风险大；这套架构的各组件均支持 “水平弹性扩容”：Flink 可通过增加 TaskManager 节点提升处理能力，Doris 可通过增加 BE 节点扩展存储与计算资源，扩容过程无需中断服务，也无需重构数据结构，能轻松应对数据量从百万到百亿的增长。

        综上，普通 MySQL 读写分离是 “保障业务交易流畅的基础方案”，而这套架构是 “挖掘数据价值、支撑实时决策的进阶方案”。在实际业务中，两者通常协同使用：MySQL 读写分离保障用户下单、支付等核心交易的稳定性，这套架构则基于交易数据实时生成分析结果，共同支撑企业 “交易高效运行 + 数据实时决策” 的双重需求。