我要评论在数据架构不断演进的背景下，从 mysql 迁移至 doris 成为许多企业提升数据处理效率的关键选择。本文将深入剖析三种经过实践验证的 mysql 迁移至 doris 的最佳方案，涵盖全量迁移、增量同步、混合迁移以及基于 cdc（change data capture）的实时迁移。每种方案都将从技术原理、实施步骤、适用场景、资源消耗等维度展开分析，并通过对比选型帮助读者结合自身业务需求与技术栈，找到最契合的迁移路径，实现数据平稳过渡与性能跃升。
一、jdbc catalog 联邦查询方案（适合跨库实时查询）
1. 方案概述
通过 doris 1.2 + 引入的jdbc catalog功能，直接通过标准 jdbc 协议连接 mysql 数据库，实现跨库联邦查询和数据写入。相比 odbc，jdbc 接口更统一、兼容性更强，且无需复杂的驱动安装和版本匹配。
2. 环境要求
| 组件 | 版本要求 | 说明 |
|---|---|---|
| doris | 1.2.0+ | 支持 jdbc catalog 功能 |
| mysql | mysql 5.7, 8.0 或更高版本 | 需开启远程访问权限 |
| mysql jdbc | 8.0.28+ | 驱动下载：mysql jdbc 驱动 |
3. 实施步骤
3.1 安装 jdbc 驱动（所有 fe/be 节点）
步骤 1：创建驱动目录
mkdir -p /your_path/doris/jdbc_drivers
步骤 2：上传驱动包下载 mysql jdbc 驱动（如mysql-connector-java-8.0.31.jar），并上传至所有 fe/be 节点的/your_path/doris/jdbc_drivers目录。
步骤 3：配置驱动路径编辑fe.conf和be.conf，添加以下配置(如果你使用的是默认的路径就不需要修改这个了)：
jdbc_drivers_dir = /opt/doris/jdbc_drivers
步骤 4：重启服务
# 重启fe cd /opt/doris/fe bin/stop_fe.sh && bin/start_fe.sh --daemon # 重启be cd /opt/doris/be bin/stop_be.sh && bin/start_be.sh --daemon
3.2 创建 jdbc catalog
使用 mysql 客户端连接 doris，执行以下 sql 创建 catalog：
create catalog mysql properties (
"type"="jdbc",
"user"="root",
"password"="secret",
"jdbc_url" = "jdbc:mysql://example.net:3306",
"driver_url" = "mysql-connector-j-8.3.0.jar",
"driver_class" = "com.mysql.cj.jdbc.driver"
)jdbc_url：mysql 连接地址，可指定数据库（如jdbc:mysql://host:port/dbname）。
driver_url：驱动文件名，需与/opt/doris/jdbc_drivers目录下的文件一致，默认就是/your_path/doris/jdbc_drivers 。
3.3 创建查询或者导入
执行联邦查询
-- 直接查询mysql表 select * from mysql_catalog.mysql_db.user_table limit 10; -- 与doris表关联查询 select a.id, a.name, b.sales from mysql_catalog.mysql_db.user_table a join doris_table b on a.id = b.user_id;
执行数据导入
-- 建表 create table as select mysql_catalog.mysql_db.user_table -- 导入 insert into doris_table select * from mysql_catalog.mysql_db.user_table
4. 注意事项
数据类型映射：
mysql 的bit类型映射为 doris 的string，需在查询时转换（如cast(bit_col as boolean)）。
大字段类型（如blob）建议通过varchar映射，避免查询性能问题。
权限控制：
为 doris 创建专用 mysql 用户，仅授予select权限，避免数据泄露。
在 doris 中通过grant语句限制外部表访问权限。
连接安全
如果您使用数据源上安装的全局信任证书配置了 tls，则可以通过将参数附加到在 jdbc_url 属性中设置的 jdbc 连接字符串来启用集群和数据源之间的 tls。
二、binlog 实时同步方案（适合增量数据同步）
1. 方案概述
通过 canal 解析 mysql binlog，实时同步增量数据至 doris，支持 insert/update/delete 操作，适用于实时数据仓库场景。
2. 环境要求
| 组件 | 版本要求 | 说明 |
|---|---|---|
| mysql | 5.7+ | 需开启 binlog（row 模式） |
| canal | 1.1.5+ | 下载地址：canal 官网 |
| doris | 1.2.0+ | 支持 batch_delete 特性 |
3. 实施步骤
3.1 配置 mysql 开启 binlog
步骤 1：编辑 my.cnf
[mysqld] log-bin=mysql-bin # 开启binlog binlog-format=row # 使用row模式 binlog-row-image=full # 记录完整行数据 server-id=1 # 唯一服务器id
步骤 2：重启 mysql 服务
sudo systemctl restart mysqld
3.2 部署 canal 服务
步骤 1：下载并解压
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz tar -zxvf canal.deployer-1.1.5.tar.gz
步骤 2：创建 instance 配置
mkdir conf/demo cp conf/example/instance.properties conf/demo/ vim conf/demo/instance.properties
关键配置：
canal.instance.master.address=192.168.1.100:3306 # mysql地址 canal.instance.dbusername=canal # 具有binlog读取权限的用户 canal.instance.dbpassword=canal # 用户密码 canal.destination=demo # instance名称
步骤 3：启动 canal
sh bin/startup.sh # 验证日志：cat logs/demo/demo.log，显示"start successful"表示启动成功
3.3 在 doris 中创建同步任务
步骤 1：创建 doris 目标表（需开启 batch_delete）
--create mysql table create table `test.source_test` ( `id` int(11) not null comment "", `name` int(11) not null comment "" ) engine=innodb default charset=utf8mb4 collate=utf8mb4_0900_ai_ci; -- create doris table create table `target_test` ( `id` int(11) not null comment "", `name` int(11) not null comment "" ) engine=olap unique key(`id`) comment "olap" distributed by hash(`id`) buckets 8 properties ( "replication_allocation" = "tag.location.default: 3" );
步骤 2：创建同步作业
create sync demo_job ( from mysql.test_table into target_test (id, name) ) from binlog ( "type" = "canal", "canal.server.ip" = "192.168.1.101", # canal服务器ip "canal.server.port" = "11111", # canal默认端口 "canal.destination" = "demo", # canal instance名称 "canal.username" = "canal", "canal.password" = "canal", "sync.mode" = "incremental" # 增量同步模式 );
4. 注意事项
权限要求：需为 canal 创建专用用户（如 canal），并授予replication slave权限：
create user 'canal'@'%' identified by 'canal'; grant replication slave on *.* to 'canal'@'%'; flush privileges;
数据一致性：建议在 doris 目标表中使用unique key 模型，并确保 mysql 源表与 doris 表字段一一对应。
三、flink cdc 流式同步方案（适合高实时性场景）
1. 方案概述
基于 flink cdc（change data capture）实时捕获 mysql 变更数据，通过 flink-doris 连接器写入 doris，支持秒级延迟，适用于实时分析、实时报表场景。
2. 环境要求
| 组件 | 版本要求 | 说明 |
|---|---|---|
| flink | 1.13+ | 支持 flink cdc 2.0+ |
| doris | 1.2.0+ | 支持 delete 操作 |
| 依赖库 | flink-cdc-2.1.0, flink-doris-connector-1.1.0 | 需下载对应版本 jar 包 |
3. 实施步骤
3.1 构建 flink 作业
步骤 1：添加 maven 依赖
<dependency> <groupid>org.apache.doris</groupid> <artifactid>flink-doris-connector-1.16</artifactid> <version>25.1.0</version> </dependency>
步骤 2：编写 flink sql 脚本
-- enable checkpoint set 'execution.checkpointing.interval' = '30s'; -- 创建mysql cdc源表 create table mysql_cdc_source ( id int primary key not enforced, name string, age int, ts timestamp(3) metadata from 'source_ts' virtual ) with ( 'connector' = 'mysql-cdc', 'hostname' = '192.168.1.100', 'port' = '3306', 'username' = 'root', 'password' = 'your_mysql_password', 'database-name' = 'test', 'table-name' = 'user_table' ); -- 创建doris目标表 create table doris_sink ( id int, name string, age int, update_time timestamp(3) ) with ( 'connector' = 'doris', 'fenodes' = 'doris-fe:8030', 'table.identifier' = 'test.user_table', 'username' = 'root', 'password' = '', 'sink.properties.format' = 'json', 'sink.enable-delete' = 'true', 'sink.label-prefix' = 'flink_cdc_' ); -- 启动数据同步 insert into doris_sink select id, name, age, ts from mysql_cdc_source;
3.2 提交 flink 作业
<flink_home>bin/flink run \
-dexecution.checkpointing.interval=10s \
-dparallelism.default=1 \
-c org.apache.doris.flink.tools.cdc.cdctools \
lib/flink-doris-connector-1.16-24.0.1.jar \
mysql-sync-database \
--database test_db \
--mysql-conf hostname=127.0.0.1 \
--mysql-conf port=3306 \
--mysql-conf username=root \
--mysql-conf password=123456 \
--mysql-conf database-name=mysql_db \
--including-tables "tbl1|test.*" \
--sink-conf fenodes=127.0.0.1:8030 \
--sink-conf username=root \
--sink-conf password=123456 \
--sink-conf jdbc-url=jdbc:mysql://127.0.0.1:9030 \
--sink-conf sink.label-prefix=label \
--table-conf replication_num=1 4. 注意事项
并行度优化：根据 mysql 表数量和数据量调整 flink 作业并行度，避免单任务压力过大。
写入机制：开启 flink checkpoint，doris flink connector 提供了两种攒批模式，默认使用基于 flink checkpoint 的流式写入方式。
| 写入方式 | 流式写入 | 批量写入 |
|---|---|---|
| 触发条件 | 依赖 flink 的 checkpoint，跟随 flink 的 checkpoint 周期写入到 doris 中 | 基于 connector 内的时间阈值、数据量阈值进行周期性提交，写入到 doris 中 |
| 一致性 | exactly-once | at-least-once，基于主键模型可以保证 exactly-once |
| 延迟 | 受 checkpoint 时间间隔限制，通常较高 | 独立的批处理机制，灵活调整 |
| 容错与恢复 | 与 flink 状态恢复完全一致 | 依赖外部去重逻辑（如 doris 主键去重） |
四、datax 同步方案（适合全量 / 批量数据迁移）
1. 方案概述
datax 是一款异构数据源之间数据同步工具，通过编写 json 格式的配置文件，实现 mysql 与 doris 之间的数据抽取、转换和加载（etl）。该方案适用于全量数据迁移、定期批量数据同步场景，对实时性要求不高，但配置灵活，可自定义数据处理逻辑。
2. 环境要求
| 组件 | 版本要求 | 说明 |
|---|---|---|
| datax | 3.0+ | 下载地址：datax 官网 |
| mysql | 5.6+ | 支持标准 jdbc 协议 |
| doris | 1.0+ | 支持通过 jdbc 或 broker 导入数据 |
3. 实施步骤
3.1 安装 datax
# 下载datax压缩包 wget https://datax-opensource.oss-cn-hangzhou.aliyuncs.com/202308/datax.tar.gz tar -zxvf datax.tar.gz
3.2 编写同步配置文件
创建mysql_to_doris.json文件，示例配置如下：
{
"job": {
"content": [
{
"reader": {
"name": "mysqlreader",
"parameter": {
"column": [xxx],
"connection": [
{
"jdbcurl": ["jdbc:mysql://localhost:3306/demo"],
"table": ["table_name"]
}
],
"username": "root",
"password": "xxxxx",
"where": ""
}
},
"writer": {
"name": "doriswriter",
"parameter": {
"loadurl": ["127.0.0.1:8030"],
"column": [xxxx],
"username": "root",
"password": "xxxxxx",
"postsql": ["select count(1) from table_name"],
"presql": [],
"flushinterval":30000,
"connection": [
{
"jdbcurl": "jdbc:mysql://127.0.0.1:9030/demo",
"selecteddatabase": "demo",
"table": ["table_name"]
}
],
"loadprops": {
"format": "json",
"strip_outer_array":"true",
"line_delimiter": "\\x02"
}
}
}
}
],
"setting": {
"speed": {
"channel": "1"
}
}
}
}3.3 执行同步任务
python bin/datax.py mysql_to_doris.json
4. 注意事项
数据类型映射：确保 mysql 与 doris 字段类型兼容，如 mysql 的datetime对应 doris 的datetime，避免类型转换错误。
性能调优：通过调整channel参数控制并发度，但过高的并发可能导致资源耗尽，建议根据数据库负载测试后设置。
五、方案对比与选型建议
| 方案 | 适用场景 | 复杂度 | 资源依赖 |
|---|---|---|---|
| jdbc catalog | 跨库实时查询、联邦分析 | ★★☆☆☆ | 依赖 doris jdbc 功能 |
| binlog 同步 | 增量数据实时同步 | ★★★☆☆ | 需部署 canal 及相关组件 |
| flink cdc 同步 | 高实时性流式处理、复杂清洗 | ★★★★☆ | 依赖 flink 集群 |
| datax 同步 | 全量 / 批量数据迁移、离线同步 | ★★★☆☆ | 轻量级工具，独立部署 |
优先选择 jdbc catalog：当需要快速验证跨库查询功能，或需要与 doris 本地表进行关联分析时。
推荐 binlog 同步：常规增量同步场景，兼顾实时性和部署复杂度。
选择 flink cdc：对实时性要求高（如实时看板），或需要复杂数据清洗（通过 flink sql 实现）。
使用 datax 同步：适合一次性全量迁移，或周期性批量同步历史数据。
五、数据迁移最佳实践要点
全量初始化与增量同步结合：
首次迁移时，先通过datax导入 mysql 全量数据，再启动 binlog 或 flink cdc 同步增量数据，避免历史数据积压。
数据类型兼容性测试：
重点验证 mysql 的enum、set、bit等类型在 doris 中的映射是否符合预期。
监控与告警体系：
结合 prometheus + grafana 监控或者doris manager工具观察资源使用情况。
这份文档已涵盖多种迁移方式。若还想对某个迁移方式补充更多细节，或增加其他方面的内容，欢迎讨论。
到此这篇关于mysql 迁移至 doris 最佳实践方案的文章就介绍到这了,更多相关mysql 迁移至 doris内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！
发表评论