当前位置: 首页 > news >正文

用 SeaTunnel 同步 MySQL 到 Doris:全量增量 + SQL 过滤

Apache SeaTunnel 能够实现 MySQL 到 Doris 的全量和增量数据同步,同时也支持 SQL 级别的数据过滤。以下是具体实现方式及功能特点:

全量与增量同步支持

1. 全量同步

实现方式:通过 SeaTunnel 的批处理模式(job.mode = "BATCH"),将 MySQL 的历史数据一次性导入 Doris。支持分片读取(如按主键分片)以提升效率,并可通过配置参数优化并行度、批量写入大小等。

source {Jdbc {query = "SELECT * FROM orders"  -- 全量数据读取partition_column = "id"        -- 分片字段split.size = 5000              -- 每分片读取行数}
}

2. 增量同步

  • 基于时间戳字段:通过WHERE update_time >= '${last_update_time}'动态参数筛选增量数据,需外部系统记录时间点并触发定期任务。
  • 基于 CDC(变更数据捕获):使用 MySQL CDC 连接器实时捕获 Binlog 变更(如插入、更新、删除),同步至 Doris。支持精确一次语义(Exactly-Once)和流式处理。
source {MySQL-CDC {startup.mode = "latest"  -- 从最新位点开始同步table-names = ["db.table"]}
}

SQL 级数据过滤支持

1. 数据源层过滤

  • 在源端 SQL 中定义过滤条件:通过 source 模块的 query 参数直接筛选数据,例如仅同步特定状态或时间范围的数据。
source {Jdbc {query = """SELECT * FROM orders WHERE status = 1 AND create_time > '2025-01-01'"""}
}
  1. 转换层过滤
    Transform 模块的 SQL 转换:在 transform 阶段通过自定义 SQL 对数据进行清洗、过滤或字段映射。
transform {Sql {query = "SELECT id, name FROM source WHERE amount > 1000"  -- 过滤金额小于1000的数据}
}

关键配置与注意事项

  1. Doris Sink 参数

需配置 Doris FE 节点地址、批量写入参数(batch_size)、数据合并策略(merge_type)等。

sink {Doris {fenodes = "doris_fe:8030"batch_size = 10000stream_load_properties = { "merge_type" = "MERGE" }}
}
  1. CDC 同步依赖条件

MySQL 需开启 Binlog 并配置ROW模式,用户需具备SELECT, REPLICATION SLAVE权限。

  1. 性能调优

全量同步建议分片读取避免单节点压力;增量同步可调整 Flink 或 Zeta 引擎的并行度以提升吞吐量。

与其他工具的对比

  • SeaTunnel 优势:支持批流一体、SQL 级灵活性、精确一次语义,适合复杂场景;相比之下,DataX 仅支持离线批量,Sqoop 局限于 Hadoop 生态 123。
  • 与 CloudCanal 对比:虽然 CloudCanal 提供可视化界面,但 SeaTunnel 的插件化架构和开源特性更适合自定义开发。

原文链接:https://blog.csdn.net/a772304419/article/details/146341445

http://www.hskmm.com/?act=detail&tid=17088

相关文章:

  • 在CodeBolcks下wxSmith的C++编程教程——使用自定义绘制和鼠标处理创建项目
  • trae 配置mysql_mcp
  • Apache NiFi 1.28.1 集群 + Kerberos 认证 + 多租户模式部署
  • 基于Java+SpringBoot+SSM,Flask福聚苑社区团购体系(源码+LW+调试文档+讲解等)/福聚苑社区/团购系统/社区团购/福聚苑/团购/社区/环境/福聚苑小区/在线团购/社区购物
  • 按需引入echarts
  • 软件构造的用户交互设计 4章
  • 自定义制作docker容器自动自愈容器镜像
  • 如何利用海外 NetNut 网络代理与 AICoding 实战获取 iPhone 17 新品用户评论数据?
  • 第一次编码器测试
  • 04-FreeRTOS的概述及编程规范
  • 10_ select/poll/epoll实现服务端的io多路复用
  • 模拟实战配置实验
  • 国标GB28181公网直播EasyGBS如何构建全域覆盖的应急管理与安全生产解决方案?
  • Serilog.AspNetCore与Serilog的区别
  • 基于MATLAB S函数实现多智能体间歇通信仿真
  • 我天,前端岗要消亡了吗?
  • 闲话
  • java8的集合新API - --
  • 基于MATLAB/Simulink的500kW三相光伏逆变器仿真
  • Docker Compose启动多个镜像实例
  • 卫星时间同步平台:助力分布式测控系统同步工作
  • mkfx 对磁盘设置标签
  • C# Avalonia 15- Animation- Easing
  • 安卓编译重点记录
  • kubelet源码阅读(二)——device plugin 的ListAndWatch过程
  • CyberLink ColorDirector Ultra 2026 14.0.5712.0 视频后期调色
  • CF 1053 Div.2
  • vi编辑器
  • 豆油
  • MQTT