简化数据流：Apache SeaTunnel实现多表同步的高效指南

我可以不吃啊  论坛元老 | 2024-7-23 14:50:53 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 1943|帖子 1943|积分 5829

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要 登录 才可以下载或查看，没有账号？立即注册

x
Apache SeaTunnel除了单表之间的数据同步之外，也支持单表同步到多表，多表同步到单表，以及多表同步到多表，下面简朴举例说明如何实现这些功能。
单表 to 单表

一个source，一个sink。
从mysql同步到mysql，中心不做区分
  1. env {
  2.   # You can set flink configuration here
  3.   execution.parallelism = 2
  4.   job.mode = "BATCH"
  5. }
  6. source{
  7.     Jdbc {
  8.         url = "jdbc:mysql://127.0.0.1:3306/test"
  9.         driver = "com.mysql.cj.jdbc.Driver"
  10.         connection_check_timeout_sec = 100
  11.         user = "user"
  12.         password = "password"
  13.         query = "select * from base_region"
  14.     }
  15. }
  17. transform {
  18.     # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
  19.     # please go to https://seatunnel.apache.org/docs/transform/sql
  20. }
  22. sink {
  23.   jdbc {
  24.         url = "jdbc:mysql://127.0.0.1:3306/dw"
  25.         driver = "com.mysql.cj.jdbc.Driver"
  26.         connection_check_timeout_sec = 100
  27.         user = "user"
  28.         password = "password"
  29.     query = "insert into base_region(id,region_name) values(?,?)"
  30.   }
  31. }
复制代码
执行任务
  1. ./bin/seatunnel.sh --config ./config/mysql2mysql_batch.conf
复制代码
单表 to 多表

一个source，多个sink。
从MySQL同步到MySQL，将一个用户表数据同步过去，中心通过2个sql组件分布将男性用户和女性用户分开，在sink阶段分别插入到差别的表：
  1. env {
  2.   execution.parallelism = 2
  3.   job.mode = "BATCH"
  4. }
  5. source {
  6.     Jdbc {
  7.         url = "jdbc:mysql://127.0.0.1:3306/test"
  8.         driver = "com.mysql.cj.jdbc.Driver"
  9.         connection_check_timeout_sec = 100
  10.         user = "user"
  11.         password = "password"
  12.         result_table_name="t_user"
  13.         query = "select * from t_user;"
  14.     }
  15. }
  17. transform {
  18.   Sql {
  19.     source_table_name = "t_user"
  20.     result_table_name = "t_user_nan"
  21.     query = "select id,name,birth,gender from t_user where gender ='男';"
  22.   }
  23.   Sql {
  24.     source_table_name = "t_user"
  25.     result_table_name = "t_user_nv"
  26.     query = "select id,name,birth,gender from t_user where gender ='女';"
  27.   }
  28. }
  30. sink {
  31.   jdbc {
  32.     url = "jdbc:mysql://127.0.0.1:3306/dw"
  33.     driver = "com.mysql.cj.jdbc.Driver"
  34.     connection_check_timeout_sec = 100
  35.     user = "user"
  36.     password = "password"
  37.     source_table_name = "t_user_nan"
  38.     query =  "insert into t_user_nan(id,name,birth,gender) values(?,?,?,?)"
  39.   }
  40.   jdbc {
  41.     url = "jdbc:mysql://127.0.0.1:3306/dw"
  42.     driver = "com.mysql.cj.jdbc.Driver"
  43.     connection_check_timeout_sec = 100
  44.     user = "user"
  45.     password = "password"
  46.     source_table_name = "t_user_nv"
  47.     query =  "insert into t_user_nv(id,name,birth,gender) values(?,?,?,?)"
  48.   }
  49. }
  50. ./bin/seatunnel.sh --config ./config/mysql2mysql_1n.conf
复制代码
多表 to 单表

多个source，一个sink。
如果有一张交换器使用情况表，一张路由器使用情况表，目标表是将这种数据合在一起的olap表。
表结构如下：
  1. -- dw 源表1
  2. CREATE TABLE IF NOT EXISTS ads_device_switch_performance (
  3.   `event_time` timestamp COMMENT '业务时间',
  4.   `device_id` VARCHAR(32) COMMENT '设备id',
  5.   `device_type` VARCHAR(32) COMMENT '设备类型',
  6.   `device_name` VARCHAR(128) COMMENT '设备名称',
  7.   `cpu_usage` INT COMMENT 'CPU使用率百分比'
  8. ) ;
  10. INSERT INTO `ads_device_switch_performance` VALUES ('2024-01-15 14:25:11', '2001', '2', '交换器1', 49);
  11. INSERT INTO `ads_device_switch_performance` VALUES ('2024-01-17 22:25:40', '2002', '1', '交换器2', 65);
  13. -- dw 源表2
  14. CREATE TABLE IF NOT EXISTS ads_device_router_performance (
  15.   `event_time` timestamp COMMENT '业务时间',
  16.   `device_id` VARCHAR(32) COMMENT '设备id',
  17.   `device_type` VARCHAR(32) COMMENT '设备类型',
  18.   `device_name` VARCHAR(128) COMMENT '设备名称',
  19.   `cpu_usage` INT COMMENT 'CPU使用率百分比'
  20. );
  22. INSERT INTO `ads_device_router_performance` VALUES ('2024-01-17 21:23:22', '1001', '1', '路由器1', 35);
  23. INSERT INTO `ads_device_router_performance` VALUES ('2024-01-16 17:23:53', '1002', '2', '路由器2', 46);
  26. -------------------------------------------------------------------------------
  27. -- olap 目标表
  28. CREATE TABLE `device_performance` (
  29.   `id` INT NOT NULL AUTO_INCREMENT COMMENT '表主键',
  30.   `event_time` VARCHAR(32) NOT NULL COMMENT '业务时间',
  31.   `device_id` VARCHAR(32) COMMENT '设备id',
  32.   `device_type` VARCHAR(32) COMMENT '设备类型',
  33.   `device_name` VARCHAR(128) NOT NULL COMMENT '设备名称',
  34.   `cpu_usage` FLOAT NOT NULL COMMENT 'CPU利用率单位是%',
  35.   `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  36.   `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '更新时间',
  37.   PRIMARY KEY (`id`)
  38. ) COMMENT='设备状态';
复制代码
将交换器数据和路由器数据一起同步到olap目标表，总结通过sql组件处理：
  1. env {
  2.     job.mode="BATCH"
  3.     job.name="device_performance"
  4. }
  6. source {
  7.     Jdbc {
  8.         url="jdbc:mysql://127.0.0.1:3306/dw?allowMultiQueries=true&characterEncoding=utf-8"
  9.         driver="com.mysql.cj.jdbc.Driver"
  10.         user = "user"
  11.         password = "password"
  12.         result_table_name="switch_src"
  13.         query="SELECT `event_time`, `device_id`, `device_type`, `device_name`, `cpu_usage` FROM ads_device_switch_performance;"
  14.     }
  15.     Jdbc {
  16.         url="jdbc:mysql://127.0.0.1:3306/dw?allowMultiQueries=true&characterEncoding=utf-8"
  17.         driver="com.mysql.cj.jdbc.Driver"
  18.         user = "user"
  19.         password = "password"
  20.         result_table_name="router_src"
  21.         query="SELECT `event_time`, `device_id`, `device_type`, `device_name`, `cpu_usage` FROM ads_device_router_performance;"
  22.     }
  23. }
  25. transform {
  26.   Sql {
  27.     source_table_name = "switch_src"
  28.     result_table_name = "switch_dst"
  29.     query = "SELECT  event_time , device_id, device_type, device_name, cpu_usage, NOW() AS create_time, NOW() AS update_time  FROM switch_src;"
  30.   }
  31.   Sql {
  32.     source_table_name = "router_src"
  33.     result_table_name = "router_dst"
  34.     query = "SELECT event_time, device_id, device_type, device_name, cpu_usage, NOW() AS create_time, NOW() AS update_time FROM router_src;"
  35.   }
  36. }
  38. sink {
  39.     Jdbc {
  40.         url="jdbc:mysql://127.0.0.1:3306/olap?allowMultiQueries=true&characterEncoding=utf-8"
  41.         driver="com.mysql.cj.jdbc.Driver"
  42.         user = "user"
  43.         password = "password"
  44.         source_table_name = "switch_dst"
  45.         query="INSERT INTO device_performance  VALUES(null,?, ?, ?, ?, ?, ?, ?) ;"
  46.       }
  47.     Jdbc {
  48.         url="jdbc:mysql://127.0.0.1:3306/olap?allowMultiQueries=true&characterEncoding=utf-8"
  49.         driver="com.mysql.cj.jdbc.Driver"
  50.         user = "user"
  51.         password = "password"
  52.         source_table_name = "router_dst"
  53.         query="INSERT INTO device_performance  VALUES(null,?, ?, ?, ?, ?, ?, ?) ;"
  54.        }
  55. }
复制代码
执行任务：
  1. ./bin/seatunnel.sh --config ./syn_job/mysql2mysql_n1_batch.conf
复制代码
作业成功！
多表 to 多表

多个source，多个sink。
将交换器使用情况数据和路由器使用情况数据分别同步到对应的目标表，中心sql组件处理
  1. env {
  2.     job.mode="BATCH"
  3.     job.name="device_performance"
  4. }
  6. source {
  7.     Jdbc {
  8.         url="jdbc:mysql://127.0.0.1:3306/dw?allowMultiQueries=true&characterEncoding=utf-8"
  9.         driver="com.mysql.cj.jdbc.Driver"
  10.         user = "user"
  11.         password = "password"
  12.         result_table_name="switch_src"
  13.         query="SELECT `event_time`, `device_id`, `device_type`, `device_name`, `cpu_usage` FROM ads_device_switch_performance;"
  14.     }
  15.     Jdbc {
  16.         url="jdbc:mysql://127.0.0.1:3306/dw?allowMultiQueries=true&characterEncoding=utf-8"
  17.         driver="com.mysql.cj.jdbc.Driver"
  18.         user = "user"
  19.         password = "password"
  20.         result_table_name="router_src"
  21.         query="SELECT `event_time`, `device_id`, `device_type`, `device_name`, `cpu_usage` FROM ads_device_router_performance;"
  22.     }
  23. }
  25. transform {
  26.   Sql {
  27.     source_table_name = "switch_src"
  28.     result_table_name = "switch_dst"
  29.     query = "SELECT  event_time , device_id, device_type, device_name, cpu_usage, NOW() AS create_time, NOW() AS update_time  FROM switch_src;"
  30.   }
  31.   Sql {
  32.     source_table_name = "router_src"
  33.     result_table_name = "router_dst"
  34.     query = "SELECT event_time, device_id, device_type, device_name, cpu_usage, NOW() AS create_time, NOW() AS update_time FROM router_src;"
  35.   }
  36. }
  38. sink {
  39.     Jdbc {
  40.         url="jdbc:mysql://127.0.0.1:3306/olap?allowMultiQueries=true&characterEncoding=utf-8"
  41.         driver="com.mysql.cj.jdbc.Driver"
  42.         user = "user"
  43.         password = "password"
  44.         source_table_name = "switch_dst"
  45.         query="INSERT INTO device_performance_switch  VALUES(null,?, ?, ?, ?, ?, ?, ?) ;"
  46.       }
  47.     Jdbc {
  48.         url="jdbc:mysql://127.0.0.1:3306/olap?allowMultiQueries=true&characterEncoding=utf-8"
  49.         driver="com.mysql.cj.jdbc.Driver"
  50.         user = "user"
  51.         password = "password"
  52.         source_table_name = "router_dst"
  53.         query="INSERT INTO device_performance_router  VALUES(null,?, ?, ?, ?, ?, ?, ?) ;"
  54.        }
  55. }
复制代码
结语

综上所述，Apache SeaTunnel多表同步技能具有高效、实时、可靠和灵活的特点，在企业的数据同步领域发挥偏重要作用。借助Apache SeaTunnel多表同步功能，企业可以或许更好地实现差别体系和数据库之间数据的无缝流转，提升数据管理和使用的效率，为业务发展提供有力支持。渴望本文可以或许帮助读者更好地了解和应用Apache SeaTunnel多表同步，从而为企业数据同步带来更多大概性。
原文链接：https://blog.csdn.net/weixin_44586883/article/details/136049897
本文由 白鲸开源 提供发布支持！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

我可以不吃啊

论坛元老
这个人很懒什么都没写!

楼主热帖

标签云

集成商 AI 运维 CIO 存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表