系列文章目录

文章目录

系列文章目录
前言

前言

玩转SpringBoot，像搭积木一样简单。在业务系统升级时，数据迁移是必不可少的一环。数据迁移又分离线和在线的方式，即存量数据和增量数据在平滑升级中融合。常用的DB数据迁移方式也有很多，文件拷贝、mysql dump、主从同步、canal、DataX、愚公、DTS等等，但这些方式往往是以镜像方式迁移，虽然有些工具支持数据类型或者字段名称转换，但支持不了复杂的转换，例如json转换等。本文将介绍如何封装支持复杂类型的数据迁移工具。
在这里插入图片描述

一、本文要点

前面的文章，介绍了基于Jolt如何实现业务型的ETL工具，本文将介绍这款工具的实际应用场景，如何快速将DB表数据迁移到另外一个DB表。系列文章完整目录

JSON to JSON
JSON 结构转换
Kafka 转存到 ES
Kafka 转存到 DB
DB 数据转KAFKA消息
DB 数据转存ES
低代码平台
SpringBoot积木

二、开发环境

jdk 1.8
maven 3.6.2
springboot 2.4.3
Jolt 0.1.5
mysql 5.6
es 6.8.2
idea 2020

三、容器模式

1、新建容器
使用制作好的etl镜像，创建工作负载，参考文章：
《搭建大型分布式服务（二十五）如何将应用部署到TKE容器集群？》
《搭建大型分布式服务（二十七）如何通过Coding流水线CI/CD将SpringBoot服务部署到TKE容器集群》
在这里插入图片描述
其中SPRING_EXT变量可以自定义profile的值，指定加载applicationn-xxx.properties文件的配置。
参考：《搭建大型分布式服务（三）SpringBoot多环境配置》

2、配置同步参数
（1）可以直接在Apollo配置中心直接修改profile对应环境的application.properties文件配置信息。
参考：《搭建大型分布式服务（十七）SpringBoot 配置托管到Apollo》
（2）也可以直接利用SpringBoot配置文件加载顺序原因，将文件application-xxx.properties挂载到容器config/application-xxx.properties目录，覆盖容器目录中的文件。


################### 共同 ###################
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.type=com.zaxxer.hikari.HikariDataSource

################### 输入 ###################
spring.jolt.input.type=db
spring.jolt.input.db.objectId=xxx  ## 可以作为入参，选择某类数据
spring.jolt.input.db.range-sql=select min(id) startIndex, max(id) endIndex from 表 where 1=1 and  objectId = ? 
spring.jolt.input.db.select-sql=select * from 表 where 1=1 and objectId = ? and id >= ? and id < ? 
spring.jolt.input.db.hikari.jdbc-url=jdbc:mysql://127.0.0.13306/xxx?useUnicode=true&characterEncoding=utf8&autoReconnect=true&useSSL=false&zeroDateTimeBehavior=convertToNull&allowMultiQueries=true&serverTimezone=GMT%2B8
spring.jolt.input.db.hikari.username=root
spring.jolt.input.db.hikari.password=
spring.jolt.input.db.hikari.minimum-idle=1
spring.jolt.input.db.hikari.maximum-pool-size=15
spring.jolt.input.db.hikari.auto-commit=true
spring.jolt.input.db.hikari.idle-timeout=30000
spring.jolt.input.db.hikari.pool-name=primary_pool
spring.jolt.input.db.hikari.max-lifetime=1800000
spring.jolt.input.db.hikari.connection-timeout=30001
spring.jolt.input.db.hikari.connection-test-query=SELECT 1 FROM DUAL
spring.jolt.input.db.hikari.connection-init-sql=set names utf8mb4

################### 转换 ###################
spring.jolt.spec.value=[{"operation":"shift","spec":{"id":"ID"}}]  ## mysql行数据转换逻辑

################### 输出 ###################
spring.jolt.output.db.enabled=true
spring.jolt.output.db.hikari.jdbc-url=jdbc:mysql://127.0.0.1:3306/xxx?useUnicode=true&characterEncoding=utf8&autoReconnect=true&useSSL=false&zeroDateTimeBehavior=convertToNull&allowMultiQueries=true&serverTimezone=GMT%2B8
spring.jolt.output.db.hikari.username=root
spring.jolt.output.db.hikari.password=
spring.jolt.output.db.hikari.minimum-idle=1
spring.jolt.output.db.hikari.maximum-pool-size=15
spring.jolt.output.db.hikari.auto-commit=true
spring.jolt.output.db.hikari.idle-timeout=30000
spring.jolt.output.db.hikari.pool-name=primary_pool
spring.jolt.output.db.hikari.max-lifetime=1800000
spring.jolt.output.db.hikari.connection-timeout=30001
spring.jolt.output.db.hikari.connection-test-query=SELECT 1 FROM DUAL
spring.jolt.output.db.hikari.connection-init-sql=set names utf8mb4
spring.jolt.output.db.table-name=新表名
spring.jolt.output.db.update-enabled=false ## 如果目标数据已存在，是否更新
spring.jolt.output.db.ignore-sinker-error=true ## 是否忽略错误

## 二选一

# 自定义语句
spring.jolt.output.db.custom-sql=insert into person (stuName, stuAge, stuSex, address) values(#{stuName}, #{stuAge}, #{stuSex}, #{address})  ## 支持mybatis语法
# 自动生成语句
spring.jolt.output.db.table-name=Person  ## 新表名
spring.jolt.output.db.include-fields=stuAge,stuSex  ## 忽略更新哪些字段
spring.jolt.output.db.update-enabled=true  ## 是否支持 on duplicate key update

################### 监控 ###################
spring.jolt.monitor.debug-field-name[0]=ID  ## 日志里打印那一列
spring.jolt.monitor.debug-pair=ID:1523655022708244503 ## 符合这个条件才打印

spring.jolt.spec.value 是基于jolt转换逻辑的核心配置项，在这可以不用写任何java代码就可以实现db数据迁移打新db表了。

3、完成配置后，启动容器，db数据迁移任务就已经正常运行。其中spring.jolt.input.db.range-sql配置控制扫表的条件，spring.jolt.input.db.select-sql控制从db里打捞数据的格式。