基于 Docker Compose 搭建 MySQL→Flink CDC→Doris 实时同步链路实战指南

基于 Docker Compose 搭建 MySQL→Flink CDC→Doris 实时同步链路实战指南【免费下载链接】dorisApache Doris is an easy-to-use, high performance and unified analytics database.项目地址: https://gitcode.com/gh_mirrors/dori/doris导读本文基于 docker/runtime/docker-compose-demo/example/mysql-flink-doris/Readme.md 及其配套的 Compose 编排与初始化脚本完整讲解如何在本地通过 Docker Compose 一键拉起 MySQL、FlinkJobManager/TaskManager与 DorisFE/BE五节点环境并借助 Flink MySQL CDC Connector 将 MySQL 的订单变更实时同步写入 Doris。读完本文你将掌握三个 Connector JAR 的挂载方式、桥接 / Host 两种网络模式的选择依据、MySQL-init → Doris-init → Flink-init三步初始化顺序以及一条可复制的实时入仓 SQL 链路。一、示例全景一条 MySQL → Flink → Doris 的实时数据链路mysql-flink-doris示例要解决的问题非常典型业务数据落在 MySQL分析场景需要同步到 Doris 的 OLAP 表中。它的整体拓扑如下MySQL 8.0.31 (orders 表) │ 通过 MySQL CDC 读取变更日志 ▼ Flink 1.15 (JobManager TaskManager加载 flink-doris-connector 与 mysql-cdc connector) │ 通过 Doris Connector 写入 ▼ Doris 2.0.0 (FE BEtest.mysql_order 表)整个示例由 5 个文件构成均在仓库 example/mysql-flink-doris 目录下文件作用Readme.md部署要点说明JAR 挂载、网络模式、启动顺序docker-compose.yaml五节点 Compose 编排mysql-init-sql.sql初始化 MySQL 库表与种子数据doris-init-sql.sql在 Doris 中建库建表目标表flink-init-sql.sqlFlink SQL定义 CDC 源表、Doris 结果表并提交同步作业仓库中同一目录下还有一个姊妹示例 hdfs-broker-doris其 Compose 与初始化脚本结构一致同样定义FE_SERVERS、BE_ADDR与固定网段172.20.80.0/24可作为理解本示例网络配置的参照。二、准备工作Connector JAR 包下载与目录映射Readme 第一条明确要求docker-compose.yaml中使用的Flink-Doris-Connector、Flink-MySQL-ConnectorJAR 包需要自行下载然后按目录进行映射。仓库不内置、也不会自动下载这两个 JAR这是运行前必须手动完成的一步。在 docker-compose.yaml 中JobManager 与 TaskManager 均通过 volume 把宿主机上的 JAR 映射进 Flink 的/opt/flink/libjobmanager: image: apache/flink:1.15-scala_2.12 volumes: - /data/flink/job/flink-doris-connector-1.15-1.2.1.jar:/opt/flink/lib/flink-doris-connector-1.15-1.2.1.jar - /data/flink/job/flink-sql-connector-mysql-cdc-2.2.1.jar:/opt/flink/lib/flink-sql-connector-mysql-cdc-2.2.1.jar taskmanager: image: apache/flink:1.15-scala_2.12 volumes: - /data/flink/task/flink-doris-connector-1.15-1.2.1.jar:/opt/flink/lib/flink-doris-connector-1.15-1.2.1.jar - /data/flink/task/flink-sql-connector-mysql-cdc-2.2.1.jar:/opt/flink/lib/flink-sql-connector-mysql-cdc-2.2.1.jar三个实操要点JAR 版本与 Flink 版本强相关本示例 Flink 镜像为apache/flink:1.15-scala_2.12对应挂载flink-doris-connector-1.15-1.2.1.jar即面向 Flink 1.15 的 Doris Connector 1.2.1 版本MySQL CDC 使用flink-sql-connector-mysql-cdc-2.2.1.jar。JobManager 与 TaskManager 都要挂两个服务分别挂载示例中宿主机路径/data/flink/job/与/data/flink/task/不同因为 Flink SQL 作业在 TaskManager 上实际执行而lib目录中的 JAR 需要保持一致。宿主机目录可自定示例中的/data/flink/job、/data/flink/task是占位路径请替换为你实际下载 JAR 后存放的目录并确保路径存在且对 Docker 守护进程可见。三、网络模式桥接bridge与 Host 两种选择Readme 第二条指出当前版本既可以用网络桥接模式也可以用 host 模式配置网络二者需要做参数调整。本示例的docker-compose.yaml采用的就是桥接模式 自定义子网 固定 IPfe: environment: - FE_SERVERSfe1:172.20.80.1:9010 - FE_ID1 networks: doris_net: ipv4_address: 172.20.80.1 be: environment: - FE_SERVERSfe1:172.20.80.1:9010 - BE_ADDR172.20.80.2:9050 networks: doris_net: ipv4_address: 172.20.80.2 networks: doris_net: ipam: config: - subnet: 172.20.80.0/24桥接模式下需要调整的核心参数FE_SERVERS格式为fe角色名:FE_IP:FE_EDIT_LOG_PORT示例为fe1:172.20.80.1:9010其中9010是 FE 的 edit log 端口该 IP 必须与networks.doris_net.ipv4_address中分配给 FE 的固定 IP 保持一致。FE_IDFE 节点的编号单 FE 部署时为1。BE_ADDR格式为BE_IP:BE_HEARTBEAT_PORT示例为172.20.80.2:9050其中9050是 BE 向 FE 汇报心跳的端口同样必须与 BE 容器在doris_net中的固定 IP 保持一致。子网约束172.20.80.0/24子网下 FE 占用.1、BE 占用.2。如果宿主机网络与该网段冲突或你改用 Host 模式都需要同步修改环境变量中的 IP。如果改用Host 模式则需要调整去掉networks、ipv4_address与自定义doris_net网络定义将FE_SERVERS中的172.20.80.1与BE_ADDR中的172.20.80.2改为宿主机实际可达的 IP注意 Host 模式下端口直接占用宿主机端口docker-compose.yaml中mysql服务的3306:3306、jobmanager的8081:8081映射方式需相应调整避免与宿主机已有进程冲突。另外mysql、jobmanager、taskmanager三个服务在示例中未加入doris_net它们之间的联通尤其是 Flink 访问 MySQL、Doris依赖默认网络与端口映射实际部署时请确保各容器网络互通例如 Flink SQL 中的${MySQL_CIP}指向可访问 MySQL 的地址。四、服务编排逐个拆解完整编排文件 定义了 5 个服务下面逐一说明其关键配置4.1 MySQL 8.0.31数据源mysql: image: mysql:8.0.31 volumes: - /data/mysql/:/var/lib/mysql environment: MYSQL_ROOT_PASSWORD: 123456 MYSQL_DATABASE: doris-test MYSQL_PASSWORD: 123456 command: --character-set-serverutf8 hostname: mysql ports: - 3306:3306数据目录持久化到宿主机/data/mysql设置 root 密码123456并预创建数据库doris-test注意CDC 实际读取的是初始化脚本中的test库MYSQL_DATABASE仅为镜像启动时自动创建的库显式指定 UTF-8 字符集避免中文字段乱码对外暴露3306端口供 Flink 及外部客户端访问。4.2 Flink JobManager / TaskManager同步引擎jobmanager: image: apache/flink:1.15-scala_2.12 expose: - 6123 ports: - 8081:8081 command: jobmanager environment: - JOB_MANAGER_RPC_ADDRESSjobmanager taskmanager: image: apache/flink:1.15-scala_2.12 expose: - 6121 - 6122 depends_on: - jobmanager command: taskmanager links: - jobmanager:jobmanager environment: - JOB_MANAGER_RPC_ADDRESSjobmanager二者共用 Flink 1.15Scala 2.12镜像JobManager 暴露8081Web UI并 expose RPC 端口6123TaskManager expose6121/6122数据交换与 RPC 端口通过JOB_MANAGER_RPC_ADDRESSjobmanager与links让 TaskManager 能解析到 JobManager 的服务名两个容器都挂载了第二节所述的 Connector JAR。4.3 Doris FE / BE目标分析库fe: image: apache/doris:2.0.0_alpha-fe-x86_64 volumes: - /data/apache-doris/fe/doris-meta/:/opt/apache-doris/fe/doris-meta - /data/apache-doris/fe/log/:/opt/apache-doris/fe/log environment: - FE_SERVERSfe1:172.20.80.1:9010 - FE_ID1 restart: on-failure be: image: apache/doris:2.0.0_alpha-be-x86_64 volumes: - /data/apache-doris/be/storage/:/opt/apache-doris/be/storage - /data/apache-doris/be/log/:/opt/apache-doris/be/log environment: - FE_SERVERSfe1:172.20.80.1:9010 - BE_ADDR172.20.80.2:9050 depends_on: - fe restart: on-failureFE 元数据与日志分别持久化到/data/apache-doris/fe/doris-meta与/data/apache-doris/fe/logBE 数据存储与日志持久化到/data/apache-doris/be/storage与/data/apache-doris/be/logdepends_on: fe保证 FE 先启动restart: on-failure提升故障恢复能力镜像为2.0.0_alpha系列注意x86_64后缀意味着需要 x86_64 架构宿主机。五、三步初始化MySQL-init → Doris-init → Flink-initReadme 第三条明确了启动顺序容器全部拉起后依次执行MySQL-init、Doris-init、Flink-init三个初始化步骤。顺序不能颠倒因为 Doris 的建表依赖 MySQL 端已有明确的表结构定义而 Flink 提交同步作业又依赖 MySQL 与 Doris 两侧的库表都已就绪。第一步MySQL-init初始化源库执行 mysql-init-sql.sql创建test库、orders表并插入 3 条种子订单数据CREATE DATABASE test; USE test; DROP TABLE IF EXISTS orders; CREATE TABLE orders ( order_id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY, order_date DATETIME NOT NULL, customer_name VARCHAR(255) NOT NULL, price DECIMAL(10, 5) NOT NULL, product_id INTEGER NOT NULL, order_status BOOLEAN NOT NULL -- Whether order has been placed ) AUTO_INCREMENT 10001; INSERT INTO orders VALUES (default, 2020-07-05 10:08:22, Jark, 50.50, 102, false), (default, 2020-07-05 10:11:09, Sally, 15.00, 105, false), (default, 2020-07-05 12:00:30, Edward, 25.25, 106, false);字段设计要点order_id自增主键CDC 依赖主键识别行变更、order_status BOOLEAN注释说明其语义为“订单是否已下单”。种子数据让 Flink 作业启动后能立刻读到历史数据便于验证同步结果。第二步Doris-init初始化目标表执行 doris-init-sql.sql在 Doris 中创建test库与目标表mysql_orderCREATE DATABASE test; USE test; DROP TABLE IF EXISTS mysql_order; CREATE TABLE test.mysql_order ( order_id INT, order_date DATETIME, customer_name VARCHAR(255), price DECIMAL(10, 5), product_id INT, order_status BOOLEAN ) UNIQUE KEY(order_id) DISTRIBUTED BY HASH(order_id) BUCKETS 3 PROPERTIES ( replication_num 1 );模型选择说明UNIQUE KEY(order_id)Doris 唯一键模型。因为同步链路需要支持对 MySQL 端已有订单的更新UPDATE与删除DELETE唯一键模型配合 Flink 侧的sink.enable-delete才能保证按主键覆盖、删除的语义正确DISTRIBUTED BY HASH(order_id) BUCKETS 3按order_id哈希分桶到 3 个分桶replication_num 1副本数设为 1。这是单节点 Docker 演示环境的必要设置——BE 只有 1 个若保持默认 3 副本会导致建表因副本不足而失败。第三步Flink-init提交实时同步作业执行 flink-init-sql.sql它由三部分组成① 定义 MySQL CDC 源表CREATE TABLE mysql_source ( order_id INT, order_date TIMESTAMP(0), customer_name STRING, product_id INT, order_status BOOLEAN, PRIMARY KEY (order_id) NOT ENFORCED ) WITH ( connector mysql-cdc, hostname ${MySQL_CIP}, port 3306, username root, password 123456, database-name test, table-name orders );connector mysql-cdc启用 MySQL CDC即依赖flink-sql-connector-mysql-cdc-2.2.1.jar${MySQL_CIP}是占位变量执行前需要替换为 MySQL 容器的实际可访问地址如容器服务名mysql或宿主机 IPPRIMARY KEY (order_id) NOT ENFORCED声明主键供 CDC 去重、识别变更NOT ENFORCED表示 Flink 侧不做强制校验。② 设置 Checkpoint 并定义 Doris 结果表SET execution.checkpointing.interval 10s; CREATE TABLE doris_sink ( order_id INT, order_date TIMESTAMP(0), customer_name VARCHAR(255), price DECIMAL(10, 5), product_id INT, order_status BOOLEAN ) WITH ( connector doris, fenodes ${HOST_ADDR}:8030, table.identifier test.mysql_order, username root, password , sink.properties.format json, sink.properties.read_json_by_line true, sink.enable-delete true, sink.label-prefix doris_label );execution.checkpointing.interval 10s开启每 10 秒一次的 Checkpoint。Doris Connector 的写入依赖 Flink Checkpoint 保证二阶段提交语义exactly-once且sink.enable-delete需要两阶段提交机制配合因此该参数是必配项fenodesDoris FE 的 HTTP 访问地址端口8030为 FE 的 HTTP 端口${HOST_ADDR}同样是需要替换的占位变量table.identifierDoris 目标表格式为库名.表名即test.mysql_ordersink.properties.format json与read_json_by_line true指定 Stream Load 使用 JSON 格式、每行一条 JSON这是 Doris 侧批量导入基于 gensrc/proto/internal_service.proto 定义的 Stream Load 通道解析数据的必要配置sink.enable-delete true开启后CDC 产生的DELETE事件会转换为 Doris 的删除操作实现“删除同步”sink.label-prefixStream Load label 前缀用于区分任务、保证幂等。③ 提交同步作业INSERT INTO doris_sink SELECT order_id, order_date, customer_name, price, product_id, order_status FROM mysql_source;一条INSERT INTO ... SELECT即将 CDC 源表持续流式写入 Doris 结果表Flink 会持续消费 MySQL binlog 中的新增、更新、删除事件并同步到 Doris 的mysql_order表。六、验证与常见问题排查验证同步效果作业提交后在 MySQL 中执行INSERT / UPDATE / DELETE再到 Doris 中SELECT * FROM test.mysql_order对比数据也可以在 JobManager 的8081Web UI 中查看作业吞吐与 Checkpoint 状态。常见问题排查思路JAR 未生效确认两个 Connector JAR 已同时挂载到 JobManager 与 TaskManager 的/opt/flink/lib且文件名与 Flink SQL 使用的connector标识匹配mysql-cdc、doris若使用 Flink SQL Client 提交需确保提交端也能加载相应 JAR。FE/BE 起不来或心跳失败检查FE_SERVERS、FE_ID、BE_ADDR与networks.doris_net中固定 IP 是否一一对应宿主机网段与172.20.80.0/24冲突时需整体调整。建表失败单 BE 环境务必设置replication_num 1否则默认 3 副本会因副本不足报错参见 doris-init-sql.sql 的写法。连通性问题${MySQL_CIP}、${HOST_ADDR}是占位变量必须替换为容器间实际可达的地址如服务名或宿主机 IP若改用 host 网络模式还需同步调整各服务的端口占用与 IP 环境变量。数据不同步优先检查sink.enable-delete、sink.properties.format json、read_json_by_line true是否与 flink-init-sql.sql 保持一致并确认 MySQL 已开启 binlogMySQL CDC 依赖 binlog 读取变更。七、小结本文以仓库 example/mysql-flink-doris 目录为骨架完整走通了“MySQL → Flink CDC → Doris”实时链路先下载并挂载两个 Connector JAR再决定桥接 / Host 网络模式并配好FE_SERVERS、BE_ADDR等关键参数随后严格按MySQL-init → Doris-init → Flink-init顺序执行三份初始化脚本。掌握这套最小可运行示例后你可以直接在其上扩展增加更多字段与表、调整 Checkpoint 间隔、接入 Flink 的窗口聚合或参照仓库中 hdfs-broker-doris 示例追加 Broker 服务把实时链路与离线导入能力组合成更完整的数据湖仓方案。【免费下载链接】dorisApache Doris is an easy-to-use, high performance and unified analytics database.项目地址: https://gitcode.com/gh_mirrors/dori/doris创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考