当有多个数据库中的一张表具有类似的数据，例如有多个不同的系统来源，数据插入不同数据库中的表。此时，若需要把这些表数据汇总成一张表进行统计分析，会面临一个棘手问题：由于多张表具有相同的主键，直接汇总到一起会出现主键重复。而 Apache SeaTunnel 就能巧妙应对这一挑战。本篇文章将详细阐述 Apache SeaTunnel 是如何解决此问题的。

本方案旨在将分布在两个独立MySQL数据库（datasource1和datasource2）中的test表，实时、准确地同步汇总到第三个数据库（datasource3）的test表中。为实现实时同步与数据修改的捕获，采用基于MySQL二进制日志（Binlog）的变更数据捕获（CDC）技术。为解决源表自增主键ID可能重复的问题，目标表将引入“数据来源”字段，并与原ID组成联合主键，确保数据的唯一性与可追溯性。

mysql> show variables where variable_name in ('log_bin', 'binlog_format', 'binlog_row_image', 'gtid_mode', 'enforce_gtid_consistency');+--------------------------+-------+| Variable_name            | Value |+--------------------------+-------+| binlog_format            | ROW   || binlog_row_image         | FULL  || enforce_gtid_consistency | OFF   || gtid_mode                | OFF   || log_bin                  | ON    |+--------------------------+-------+


如果log_bin的值不是ON, 配置Mysql配置文件mysql.cnf。

log-bin=mysql-binserver-id=1binlog_format=ROWbinlog_checksum=NONE


${SEATUNNEL_HOME}/config/plugin_config文件,增加

我是用的是mysql-connector-java-8.0.28.jar

把jar包放到${SEATUNNEL_HOME}/lib/目录下

CREATE DATABASE source1CHARACTER SET utf8mb4;CREATE DATABASE source2CHARACTER SET utf8mb4;CREATE DATABASE source3CHARACTER SET utf8mb4;use source1;CREATE TABLE `test`  (  `id` int(11) NOT NULL AUTO_INCREMENT,  `name` varchar(50) CHARACTER SET utf8mb4,  PRIMARY KEY (`id`) USING BTREE) ENGINE = InnoDB AUTO_INCREMENT = 1 CHARACTER SET = utf8mb4 ;
insert into test values(1,'张三');insert into test values(2,'李四');
use source2;CREATE TABLE `test`  (  `id` int(11) NOT NULL AUTO_INCREMENT,  `name` varchar(50) CHARACTER SET utf8mb4,  PRIMARY KEY (`id`) USING BTREE) ENGINE = InnoDB AUTO_INCREMENT = 1 CHARACTER SET = utf8mb4 ;
insert into test values(1,'王五');insert into test values(2,'赵六');
use source3;CREATE TABLE `test`  (  `id` int(11) NOT NULL AUTO_INCREMENT,  `name` varchar(50) CHARACTER SET utf8mb4,  `sources` varchar(50) CHARACTER SET utf8mb4,  PRIMARY KEY (`id`, `sources`) USING BTREE) ENGINE = InnoDB AUTO_INCREMENT = 1 CHARACTER SET = utf8mb4 ;


env {  parallelism = 2  job.mode = "STREAMING"  checkpoint.interval = 20000}
source {  MySQL-CDC {    plugin_output = "source_data1"    url = "jdbc:mysql://10.0.12.100:3306/source1"    username = "root"    password = "root"    table-names = ["source1.test"]    startup.mode = initial      }  MySQL-CDC {    plugin_output = "source_data2"    url = "jdbc:mysql://10.0.12.100:3306/source2"    username = "root"    password = "root"    table-names = ["source2.test"]    startup.mode = initial      }}transform {  Sql {    plugin_input = "source_data1"    plugin_output = "result1"    query = "SELECT *, 'source1' AS sources FROM source_data1"  }  Sql {    plugin_input = "source_data2"      plugin_output = "result2"    query = "SELECT *, 'source2' AS sources FROM source_data2"  }}sink {  Jdbc {    plugin_input = ["result1","result2"]    url = "jdbc:mysql://10.0.12.100:3306/source3"    driver = "com.mysql.cj.jdbc.Driver"    username = "root"    password = "root"    database = "source3"    table = "test"    generate_sink_sql = true    primary_keys = ["id","sources"]     }}


这里使用了一个配置来同步2个库，当然也可以配置成2个任务进行同步。

这里使用2个source，一个source_data1，一个source_data2。

transform也使用2个Sql，进行处理对应的2个

source。这里写成固定值来读取sources。

sink对应上面的2个Sql，输出表test，这里设置主键用于修改。

这里目标表需要手动创建，否则提示下面错误

BLOB/TEXT column 'sources' used in key specification without a key length


bin/seatunnel.sh --config job/mysql_mysql.conf -m local


mysql> select * from test;+----+--------+---------+| id | name   | sources |+----+--------+---------+|  1 | 张三   | source1 ||  1 | 王五   | source2 ||  2 | 李四   | source1 ||  2 | 赵六   | source2 |+----+--------+---------+


use source1;insert into test values(3,'钱七');update test set name='张三1' where id=1;use source2;delete from test where id=1;


mysql> select * from test;+----+---------+---------+| id | name    | sources |+----+---------+---------+|  1 | 张三1   | source1 ||  2 | 李四    | source1 ||  2 | 赵六    | source2 ||  3 | 钱七    | source1 |+----+---------+---------+


以上操作就是把2个库中的表数据同步到另一个库中的一张表的所有过程，我有看到其他文章可以自动创建输出表，但一直没有测试出来，有测试出来的朋友可以留言相告。

点亮Star！https://github.com/apache/seatunnel

当多表数据涌入，Apache SeaTunnel 如何巧妙化解主键冲突？

数据库

大数据

后端

Apache SeaTunnel实现多源MySQL表数据合并，通过CDC技术捕获变更数据，采用联合主键解决ID冲突问题。配置简单高效，支持实时同步与数据修改追踪，轻松实现跨库表数据汇总分析。

数据同步

SQL数据库

MySQL

2026上云采购 | AI焕新·智启新局

datainlong

腾讯云数据库特惠，新用户首单0.5折起


文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

当多表数据涌入，Apache SeaTunnel 如何巧妙化解主键冲突？

当多表数据涌入，Apache SeaTunnel 如何巧妙化解主键冲突？

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐