对应db2迁移到mysql方案在网上都是使用navcat，这个种方案在生产环境不现实，因为生产环境基本上时命令行方式，所以优先想到的是使用命令行到处txt文件，然后导入到mysql，使用kettle等etl工具进行导入，最后想到手写一个java服务迁移。
使用命令行方式导入 db2的导出方式先连接db2数据库
db2 connect to 数据库名
导出命令 chardel"表示双引号为字段定界符 coldel,表示逗号为列定界符 codepage=1208表示导出编码为utf-8
db2 "export to 'xxx/xxx.txt' of del modified by chardel" coldel, codepage=1208 select * from schema.表名"
但是使用上面的命令发现大字段会进行截取，最长只能32700，同时也通过命令db2 get db cfg和db2set -all查看了数据库编码确认为utf-8，所以对应大字段导出增加一个关键字lobsinfile
db2 "export to 'xxx/xxx.txt' of del modified by chardel" coldel, codepage=1208 lobsinfile select * from schema.表名"
但是导出发现会生成一个lob文件专门存储大字段，无法导入到mysql了，所以采用shell脚本拼接大字段，要将schema替换成对应的，也可以修改成传参方式
#!/bin/bash
#创建导出目录
madir ~/back
echo "导出目录为：${HOME}/back"
#连接数据库
if ( db2 connect to '数据库名' >/dev/null )
then
echo "数据库连接成功"
else
echo "数据库连接失败"
fi
#公共变量
maxLength=0 #字段最大字符数
colStr="" #表的列拼接字符串
#获得当前列最大字符长度 参数为：表名 列名
function getMaxLength(){
sql="select max(length($2)) from ${1}"
maxLength=`db2 -x ${sql}`
#去除空格
maxLength=`eval echo "${maxLength}"`
}
#获得表的列拼接的sql查询字符串 参数：表名
function getColumnStr(){
#获取表的所以列和类型
querySql="select colname,typename from syscat.columns where tabschema = 'schema' and tabname = '${1}' order by colno"
datas=(`db2 -x ${sql}`)
colStr=""
#遍历类型 数组存储方式是下标为偶数是列名，下标为奇数是类型，所以从1开始每次步长为2
for((i=1;i<${#datas[@]};i+=2))
do
# 如果不是大字段直接拼接
if [[ "#{datas[i]}" != "CLOB" && "#{datas[i]}" != "BLOB" ]];then
colStr="${colStr},${datas[i-1]}"
continue
fi
#获得此表的此列的最大字符长度
getMaxLength "schema.${1}" "${datas[i-1]}"
column="CLOB(${datas[i-1]})"
#如果结果为-，说明值为null
if [ "${maxLength}" == '-' ];then
colStr="${colStr},${column}"
continue
fi
#计算拆分循环次数
num=`expr ${maxLength} / 32600`
if(( `expr ${maxLength} % 32600` != 0 ));then
num=`expr ${num} + 1`
fi
#开头不加标记
colStr="${colStr},rtrim(left(substr(${column},${startNum}),32600))"
#开始拼接，第二个拆分加@标记
for((j=1;j<${num};j++))
do
startNum=`expr ${i} * 32600`
temp="${temp},'@'||rtrim(left(substr(${column},${startNum}),32600))"
done
colStr="${colStr},${temp}"
done
colStr=${colStr:1}
}
function execSql(){
#获得表的查询sql
getColumnStr "${1}"
#导出为#字段定界符、$列定界符、编码utf-8
exportSql="export to '${HOME}/back/${1}.txt' of del modified by chardel# coldel$ codepage=1208 ${colStr}"
db2 "${exportSql}">/dev/null
if(( $? == 4 ));then
echo "${1}：导出失败"
fi
#替换特殊标记，将大字段连接一起
sed -i 's/#$#@//g' "${HOME}/back/${1}.txt"
sed -i 's/$$/$NULL$/g' "${HOME}/back/${1}.txt"
sed -i 's/$$/$NULL$/g' "${HOME}/back/${1}.txt"
sed -i 's/$$/$NULL/g' "${HOME}/back/${1}.txt"
}
echo "开始导出数据..."
execSql "表名"
}
mysql导入方式
mysql -u用户名 -p密码 --local-infile mysql> use 数据库; mysql> set global local_infile=1; mysql> set foreign_key_checks=0; mysql> load data local infile '导出文件' into table 表名 character set utf8mb4 fields terminated by '$' enclosed by '#' escaped by '#' lines terminated by 'n';
在进行导入的时候有部分表会报unknow error 1300，猜测有可能是字符集问题，使用gbk导入虽然不会报错，但导入的数据有问题，猜测有可能是数据包含特殊字符导致导入有差异最终放弃此方式迁移
使用ETL工具迁移虽然在生产上不能用，但为了了解在本地上试了一下，会有各种问题所以选择放弃，我也只成功一次后面就莫名失败。
使用java代码方式这个就比较简单了直接拼接sql语句插进去就行了
pom.xml
4.0.0 org.springframework.boot spring-boot-starter-parent 2.7.2 com.example db-transfer 0.0.1-SNAPSHOT db-transfer db-transfer 17 org.springframework.boot spring-boot-starter-data-jdbc com.ibm.db2 jcc runtime mysql mysql-connector-java runtime org.springframework.boot spring-boot-starter-test test org.springframework.boot spring-boot-maven-plugin
application.yml
spring:
datasource:
mysql:
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://192.168.137.137:3306/rule?serverTimezone=GMT
username: root
password: 123456
db2:
driver-class-name: com.ibm.db2.jcc.DB2Driver
url: jdbc:db2://192.168.137.137:50000/testdb:currentSchema=RULE;serverTimezone=GMT;
username: test
password: 123456
启动类DbTransferApplication.java
package com.example.dbtransfer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Primary;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.SingleConnectionDataSource;
import javax.sql.DataSource;
@SpringBootApplication
public class DbTransferApplication {
public static void main(String[] args) {
SpringApplication.run(DbTransferApplication.class, args);
}
@Bean(name = "mysql")
@Primary
@ConfigurationProperties("spring.datasource.mysql")
public DataSource mysql(){
return new SingleConnectionDataSource();
}
@Bean(name = "db2")
@ConfigurationProperties("spring.datasource.db2")
public DataSource db2(){
return new SingleConnectionDataSource();
}
@Bean(name = "mysqlJdbc")
@Primary
public JdbcTemplate mysqlJdbc(){
JdbcTemplate jdbcTemplate = new JdbcTemplate();
jdbcTemplate.setDataSource(mysql());
return jdbcTemplate;
}
@Bean(name = "db2Jdbc")
public JdbcTemplate db2Jdbc(){
JdbcTemplate jdbcTemplate = new JdbcTemplate();
jdbcTemplate.setDataSource(db2());
return jdbcTemplate;
}
}
数据库迁移类TransferDB.java
package com.example.dbtransfer.db;
import com.example.dbtransfer.page.Page;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
public class TransferDB {
// 日志打印
private Logger log = LoggerFactory.getLogger(this.getClass());
// 源数据库
private DBBuildAbstract from;
// 目标数据库
private DBBuildAbstract to;
// 类型转换函数，可以自定义类型转换
private Function format;
public DBBuildAbstract getFrom() {
return from;
}
public void setFrom(DBBuildAbstract from) {
this.from = from;
}
public DBBuildAbstract getTo() {
return to;
}
public void setTo(DBBuildAbstract to) {
this.to = to;
}
public Function getFormat() {
return format;
}
public void setFormat(Function format) {
this.format = format;
}
public TransferDB(DBBuildAbstract from, DBBuildAbstract to) {
this.from = from;
this.to = to;
// 默认直接返回，不转换
this.format = o -> o;
}
public void transBach(int num) {
// 获取源的所有表名
List tables = from.getTables();
// 遍历表名
tables.parallelStream().forEach(table -> {
try {
// 设置目标数据库外键失效
to.keyDisable(true);
//获取记录数
int count = from.getCount(table);
// 封装page进行分页查询
Page page = new Page(num, count);
for (int i = 0; i < page.getTotalPage(); i++) {
//设置当前页
page.setCurrentPage(i);
// 获取源数据库数据
List> data = from.getDataPage(table, page);
// 目标数据库获取insert语句
String insertBachSql = to.getInsertBachSql(table, data.get(0).keySet(), data.size());
// 组成参数集合
List
数据库sql构建抽象类DBBuildAbstract.java
package com.example.dbtransfer.db;
import com.example.dbtransfer.page.Page;
import org.springframework.jdbc.core.JdbcTemplate;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
public abstract class DBBuildAbstract {
// 表名或者字段名定界符
protected String enclosed;
// 数据库schema
protected String schema;
// jdbc连接执行sql
protected JdbcTemplate jdbcTemplate;
public DBBuildAbstract(String enclosed, String schema, JdbcTemplate jdbcTemplate) {
this.enclosed = enclosed;
this.schema = schema;
this.jdbcTemplate = jdbcTemplate;
}
public String getEnclosed() {
return enclosed;
}
public void setEnclosed(String enclosed) {
this.enclosed = enclosed;
}
public String getSchema() {
return schema;
}
public void setSchema(String schema) {
this.schema = schema;
}
public JdbcTemplate getJdbcTemplate() {
return jdbcTemplate;
}
public void setJdbcTemplate(JdbcTemplate jdbcTemplate) {
this.jdbcTemplate = jdbcTemplate;
}
public abstract List getTables();
public abstract void keyDisable(boolean b);
public int getCount(String table) {
Integer count = jdbcTemplate.queryForObject("select count(1) from " + enclosed + table + enclosed, int.class);
count = Objects.isNull(count) ? 0 : count;
return count;
}
;
public List> getData(String table) {
return jdbcTemplate.queryForList("select * from " + enclosed + table + enclosed);
}
public List> getDataPage(String table, Page page) {
String sql = "select * from " + enclosed + table + enclosed + " limit "
+ (page.getCurrentPage() * page.getPageSize()) + "," + page.getPageSize();
return jdbcTemplate.queryForList(sql);
}
public String getInsertSql(String table, Set columnList) {
StringBuilder columns = new StringBuilder();
StringBuilder value = new StringBuilder();
for (String key : columnList) {
columns.append(",").append(enclosed).append(key).append(enclosed);
value.append(",").append("?");
}
columns.delete(0, 1);
value.delete(0, 1).insert(0, "(").append(")");
return "insert into " + enclosed + table + enclosed + " (" +
columns + ") values " + value;
}
public String getInsertBachSql(String table, Set columnList, int rowNum) {
StringBuilder columns = new StringBuilder();
StringBuilder value = new StringBuilder();
StringBuilder values = new StringBuilder();
for (String key : columnList) {
columns.append(",").append(enclosed).append(key).append(enclosed);
value.append(",").append("?");
}
columns.delete(0, 1);
value.delete(0, 1).insert(0, "(").append(")");
for (int i = 0; i < rowNum; i++) {
values.append(",").append(value);
}
return "insert into " + enclosed + table + enclosed + " (" +
columns + ") values " + values.delete(0, 1);
}
}
mysql的sql语句构建类MysqlSqlBuild.java
package com.example.dbtransfer.db;
import org.springframework.jdbc.core.JdbcTemplate;
import java.sql.SQLException;
import java.util.List;
import java.util.Objects;
public class MysqlSqlBuild extends DBBuildAbstract {
public MysqlSqlBuild(String schema, JdbcTemplate jdbcTemplate) {
super("`",schema,jdbcTemplate);
try {
if (!(Objects.requireNonNull(jdbcTemplate.getDataSource()).getConnection()
.getMetaData().getDatabaseProductName().toLowerCase().contains("mysql"))) {
throw new RuntimeException("数据库不对应");
}
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
@Override
public List getTables() {
String sql = "select TABLE_NAME from information_schema.tables where" +
" TABLE_SCHEMA = '"+schema+"' order by TABLE_NAME";
return jdbcTemplate.queryForList(sql, String.class);
}
@Override
public void keyDisable(boolean b) {
int key = b ? 0 : 1;
jdbcTemplate.execute("SET FOREIGN_KEY_CHECKS = " + key);
}
}
db2sql语句构建类DB2SqlBuild.java
package com.example.dbtransfer.db;
import org.springframework.jdbc.core.JdbcTemplate;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;
import java.util.Objects;
public class DB2SqlBuild extends DBBuildAbstract {
public DB2SqlBuild(String schema, JdbcTemplate jdbcTemplate) {
super(""",schema,jdbcTemplate);
try {
if (!(Objects.requireNonNull(jdbcTemplate.getDataSource()).getConnection()
.getMetaData().getDatabaseProductName().toLowerCase().contains("db2"))) {
throw new RuntimeException("数据库不对应");
}
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
@Override
public List getTables() {
String sql = "select TABNAME from syscat.TABLES where TABSCHEMA = '" +
schema + "' order by TABSCHEMA";
return jdbcTemplate.queryForList(sql, String.class);
}
@Override
public void keyDisable(boolean b) {
String disableStr = b ? "NOT" : "";
String sql = "select CONSTNAME,TABNAME from syscat" +
".references where TABSCHEMA = '" + schema + "';";
List> forKeyList = jdbcTemplate.queryForList(sql);
if (forKeyList.size() == 0) return;
forKeyList.forEach(key -> {
String sql1 = "ALTER TABLE "" + key.get("TABNAME") + "" ALTER FOREIGN KEY "" +
key.get("CONSTNAME") + "" " + disableStr + " ENFORCED";
jdbcTemplate.execute(sql1);
});
}
}
测试
package com.example.dbtransfer;
import com.example.dbtransfer.db.DB2SqlBuild;
import com.example.dbtransfer.db.MysqlSqlBuild;
import com.example.dbtransfer.db.TransferDB;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.jdbc.core.JdbcTemplate;
@SpringBootTest
class DbTransferApplicationTests {
private Logger log = LoggerFactory.getLogger(this.getClass());
@Autowired
@Qualifier("mysqlJdbc")
private JdbcTemplate mysql;
@Autowired
@Qualifier("db2Jdbc")
private JdbcTemplate db2;
@Test
void testTrans(){
DB2SqlBuild db2SqlBuild = new DB2SqlBuild("RULE", db2);
MysqlSqlBuild mysqlSqlBuild = new MysqlSqlBuild("rule", mysql);
TransferDB transferDB = new TransferDB(db2SqlBuild, mysqlSqlBuild);
long l = System.currentTimeMillis();
//transferDB.trans();//15647
//transferDB.transBach(10000);//2064
log.info("用时：{}", System.currentTimeMillis() - l);
}
@Test
void testTrans2(){
DB2SqlBuild db2SqlBuild = new DB2SqlBuild("RULE", db2);
MysqlSqlBuild mysqlSqlBuild = new MysqlSqlBuild("rule", mysql);
TransferDB transferDB2 = new TransferDB(mysqlSqlBuild, db2SqlBuild);
// 自定义数据转换
transferDB2.setFormat(o->{
if (o instanceof LocalDateTime) {
return Timestamp.valueOf((LocalDateTime) o);
} else if (o instanceof LocalDate) {
return Timestamp.valueOf(LocalDateTime.of((LocalDate) o, LocalTime.MIN));
} else if (o instanceof LocalTime) {
return Timestamp.valueOf(LocalDateTime.of(LocalDate.now(), (LocalTime) o));
}
return o;
});
long l = System.currentTimeMillis();
transferDB2.trans();//1136
transferDB2.transBach(1000);//3892
log.info("用时：{}", System.currentTimeMillis() - l);
}
}
经过测试，发现java性能比命令行差了不少，但在对比数据完整性时没有问题，所有最终选择java方式。
