# DataSync
**Repository Path**: minibox/DataSync
## Basic Information
- **Project Name**: DataSync
- **Description**: 数据同步工具Oracle/DB2/DWS/OpenGauss/OtherJDBC TO DWS/OpenGauss
- **Primary Language**: Java
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No
## Statistics
- **Stars**: 2
- **Forks**: 8
- **Created**: 2022-08-24
- **Last Updated**: 2024-04-06
## Categories & Tags
**Categories**: Uncategorized
**Tags**: None
## README
---
typora-copy-images-to: ./images
---
# DataSync
#### 介绍
数据同步/表结构迁移/数据倾斜/表结构对比 分析工具
弥补目前官方工具空缺或服务器操作权限不够大问题。
##### 数据同步
1. 由任意JDBC数据库 To openGauss/GaussDB(DWS)
2. 由CSV或TEXT数据文件 To openGauss/GaussDB(DWS) To 数据文件
- 同步基于JDBC COPY接口，性能高于INSERT语句，低于GDS外表。
具有业务数据无需落地成文件的优势，运行时低内存开销即可完成数据同步。
- 常规迁移流程需要先导出数据文件，数据文件字符集转码(GBK转UTF8)，特殊字符(回车符换行符或其他Gauss数据库无法识别的字符)清洗等操作，然后再将数据文件移动到GDS服务器，创建对应外表，最后再从外表写入实际表。流程复杂人工干预过多，占用本来就紧张的测试开发服务器存储空间。
- 同步工具执行开始后，不用人工干预同步过程，自动化多线程完成跨库写入，只需最终人工确认有无同步失败的任务和同步数据量。
- 20分钟10个线程，从Oracle所在Doker容器，全库共有1600万数据，迁移到openGauss所在Doker容器，同一台服务器不同容器。
- 20分钟1个线程，从Oracle虚拟机，交易流水表111个字段一天2000万交易数据，一个分区，迁移到GaussDB(DWS)，10万数据提交一次，峰值内存使用1.8GB，跨物理服务器。
- 57分钟10个线程从GaussDB(DWS) A实例，最大的一个账户信息表1.8亿数据，其余为交易流水表等总数据量高达10亿+，迁移到 B实例 同为Gauss数据库峰值内存仅使用100MB，同一台服务器。
- 6分钟10个线程，从DB2所在Doker容器，全库共有1000万数据，迁移到openGauss所在Doker容器，同一台服务器不同容器。
##### 表结构迁移
1. Oracle 目前支持 表结构(字段,主键,索引,RANGE分区)/视图 导出成Gauss兼容的语法
2. openGauss/GaussDB(DWS) 目前支持 表结构(字段,主键,索引,分区)/视图 导出
3. 其他数据库支持 表结构(字段,主键,索引) 导出成Gauss兼容的语法
4. 支持自动生成推荐的分布列，但是建议人工确认
- GBK字符集转UTF8字符集，可以对VARCHAR类型字段进行扩长操作，支持自定义扩长倍数。
- 向GaussDB(DWS)列存储迁移表，若不指定分布列字段，会默认第一个字段作为分布列，最终可能会导致数据倾斜问题。
- 函数及存储过程，因兼容性不同，改造的地方比较多，只建议人工整理迁移。
##### 数据倾斜分析
1. 仅GaussDB(DWS)支持数据倾斜分析，原理是根据不同节点数据分布情况统计得出。
2. 不需要高级权限，仅依赖于拥有表查询权限。
- 目的是提前在测试阶段就发现数据倾斜问题，避免上线后在运行过程中被监测到才发现问题。
- 系统中有内置的视图可以查看数据倾斜问题，但是测试环境用户可能也没有高级权限，所以用该工具替代检查数据倾斜问题。
- 实现逻辑是用利用工具去查询系统中每一张表的数据分布情况，根据GaussDB推荐的人工分析实现。
- 因openGauss不是分布式节点，程序不会报错，但是无意义。
##### 表结构对比
1. 迁移到新环境可能会涉及代码逻辑修改调整，表结构可能会新增字段或者删减字段，若没有规范记录，造成环境差异，导致最终上线混乱，所以新增此功能。
2. 通过easyExcel生成Excel表格，留存记录保存。
###### GaussDB(DWS)官方说明
GaussDB(DWS)是采用Shared-nothing架构的MPP(Massive Parallel Processor，大规模并发处理)系统，采用水平分布的方式，将业务数据表的元组按合适的分布策略分散存储在所有的DN。
当前产品支持复制(Replication)和散列(Hash)两种用户表分布策略。
- Replication方式:在每一个DN上存储一份全量表数据。对于数据量比较小的表建议采取Replication分布策略。
- Hash方式:采用这种分布方式，需要为用户表指定一个分布列(distribute key)。当插入一条记录时，系统会根据分布列的值进行hash运算后，将数据存储 在对应的DN中。对于数据量比较大的表建议采取Hash分布策略。
对于Hash分布策略，如果分布列选择不当，可能导致数据倾斜。因此在采用Hash分布策略之后会对用户表的数据进行数据倾斜性检查，以确保数据在各个DN上是均匀分布的。一般情况下分布列都是选择键值重复度小，数据分布比较均匀的列。
#### 软件架构
由Jdk1.8编写，基于JDBC底层代码驱动，不涉及其他更多第三方包。
#### 安装教程
1. 需要安装Jdk1.8及以上版本
2. 准备相应数据库的jdbc驱动jar包
3. 数据直接同步,需要两个数据库能在某个环境节点能直接互通
#### 使用说明
初次执行会在jar包同级目录生成lib文件夹（请把JDBC驱动jar包放到这里）和config.properties配置文件以及功能提示：
```sh
java -jar DataSync-x.x.x.jar
```
修改config.properties以满足使用：
```properties
# 数据库配置信息
# 前缀不重要,每一个数据库配置必须包含driver/url/username/password
# 若schema用户默认的就是要导入or导出的,可以不用指定. 或者如果数据支持可以在url中对currentSchema进行设置
oracle.driver=oracle.jdbc.OracleDriver
oracle.url=jdbc:oracle:thin:@localhost:1521:amldb
oracle.username=aml
oracle.password=aml
db2.driver=com.ibm.db2.jcc.DB2Driver
db2.url=jdbc:db2://localhost:50000/amldb:currentSchema=AML;
db2.username=aml
db2.password=aml
openGauss.driver=org.postgresql.Driver
openGauss.url=jdbc:postgresql://localhost:15432/amldb
openGauss.schema=aml
openGauss.username=aml
openGauss.password=admin@db2
dws.driver=org.postgresql.Driver
dws.url=jdbc:postgresql://localhost:5432/amldb?currentSchema=aml
dws.username=aml
dws.password=aml@1234
# 以下为运行参数 建议谨慎设置
# varchar类型扩长倍数:来源库若为GBK字符集,目标库变长字符类型将扩长的倍数
create.varchar=1.5
# ORIENTATION 取值范围=COLUMN:表的数据将以列式存储。 ROW:表的数据将以行式存储。
create.orientation=ROW
# COMPRESSION 取值范围:列存表的有效值为YES/NO/LOW/MIDDLE/HIGH，默认值为LOW。 行存表不支持压缩。
# 指定表数据的压缩级别，它决定了表数据的压缩比以及压缩时间。一般来讲，压缩级别越高，压缩比也越大，压缩时间也越长;反之亦然。实际压缩比取决于加载的表数据的分布特征。
create.compression=NO
# 是否导出索引创建语句 0:否 1:是 OpenGauss建议使用 GaussDB(DWS)不建议使用
create.index=1
# 是否导出分布列 0:否 1:是 OpenGauss不支持 GaussDB(DWS)支持
create.distribute=0
# 是否导出视图 0:否 1:是
create.view=1
# 导出字段默认不能为空 0:否 1:是
create.notnull=0
# 在文件中分隔各个字段的字符串，分隔符最大长度不超过10个字节。
# 取值范围:不允许包含\.abcdefghijklmnopqrstuvwxyz0123456789中的任何一个 字符。
sync.delimiter=\u007f\u005e
# 指定数据文件的编码格式名称
sync.encoding=UTF8
# 指定导入导出数据文件换行符样式。
# 取值范围:支持多字符换行符，但换行符不能超过10个字节。常见的换行 符，如\r、\n、\r\n(设成\u000D、\u000A、\u000D\u000A效果是相同的)，其他字符 或字符串，如$、#。
sync.eol=\u0001\u0003
# 用来指定数据文件中空值的表示。 取值范围:
# null值不能是\r和\n，最大为100个字符。 null值不能和分隔符参数相同
sync.null=
# 从其他数据库同步时提交条数
sync.commit=100000
# 同步线程数
sync.pool=10
# 会话超时时间 (分钟)
sync.timeout=180
```
##### 同步表
1. 从某个Gauss数据源，将某个表数据卸载到服务器指定路径（后面的参数顺序不重要）
```sh
java -jar DataSync-x.x.x.jar -f 绝对路径 -o 数据源名称 -t 表名
```
2. 从服务器上的数据文件，加载到某个Gauss数据源的某个表
```sh
java -jar DataSync-x.x.x.jar -f 绝对路径 -i 数据源名称 -t 表名
```
3. 两个GaussDB直接指定表名进行同步
```sh
java -jar DataSync-x.x.x.jar -o 写出数据源名 -i 写入数据源名 -t 表名
```
4. 两个GaussDB根据SQL确定数据范围进行同步
```sh
java -jar DataSync-x.x.x.jar -o 写出数据源名 -i 写入数据源名 -s "在写出数据源执行的SQL查询语句" -t 写入表名
```
5. 从任意数据库向GaussDB同步数据，初次以这种模式执行，若写入数据源不存在T18_SYNC的表，那么会生成init+时间戳的sql文件，需要在写入数据源执行。
```sh
java -jar DataSync-x.x.x.jar -o 写出数据源名 -i 写入数据源名
```
主要分为建立T18_SYNC表 和 根据GaussDB 用户下的表创建初始化任务，sql文件内容如下：
```sql
/* 在目标库 创建T18_SYNC同步任务表 */
DROP TABLE IF EXISTS aml.T18_SYNC;
CREATE TABLE IF NOT EXISTS aml.T18_SYNC
(
SYNC_SQL VARCHAR(4000) NOT NULL,
PRE_SQL VARCHAR(4000),
BEGIN_DATE TIMESTAMP,
END_DATE TIMESTAMP,
STATUS CHAR(1) NOT NULL,
ERROR_MSG VARCHAR(4000),
SYNC_DATA_SIZE INTEGER,
LANDING_TABLENAME VARCHAR(32),
LANDING_SCHEMA VARCHAR(32),
CONSTRAINT PK_T18_SYNC PRIMARY KEY (SYNC_SQL)
);
COMMENT ON TABLE aml.T18_SYNC IS '同步任务表';
COMMENT ON COLUMN aml.T18_SYNC.SYNC_SQL IS '在写出数据源的执行语句:表名(表示同步全字段全表);表名+WHERE(表示按条件同步全字段);SELECT 字段名 FROM 表名(选择字段同步,也可以加WHERE条件)';
COMMENT ON COLUMN aml.T18_SYNC.PRE_SQL IS '在写入数据源执行的前置处理SQL,一般用于数据导入前的清理语句';
COMMENT ON COLUMN aml.T18_SYNC.BEGIN_DATE IS '同步开始时间';
COMMENT ON COLUMN aml.T18_SYNC.END_DATE IS '同步结束时间';
COMMENT ON COLUMN aml.T18_SYNC.STATUS IS '状态:0同步失败;1待同步;2同步中;3同步完成';
COMMENT ON COLUMN aml.T18_SYNC.ERROR_MSG IS '同步错误信息';
COMMENT ON COLUMN aml.T18_SYNC.SYNC_DATA_SIZE IS '同步数据条数';
COMMENT ON COLUMN aml.T18_SYNC.LANDING_TABLENAME IS '落地表名';
COMMENT ON COLUMN aml.T18_SYNC.LANDING_SCHEMA IS '落地模式名';
/* 向目标库 插入全库同步作业 */
INSERT INTO aml.T18_SYNC (SYNC_SQL, PRE_SQL, STATUS, LANDING_SCHEMA)
SELECT CASE
WHEN T1.RELNAME IS NULL THEN T.RELNAME
ELSE T.RELNAME || ' PARTITION (' || T1.RELNAME || ')'
END SYNC_SQL,
CASE
WHEN T1.RELNAME IS NULL THEN 'TRUNCATE TABLE ' || T2.SCHEMANAME || '.' || T.RELNAME
ELSE 'ALERT TABLE ' || T2.SCHEMANAME || '.' || T.RELNAME || ' TRUNCATE PARTITION ' || T1.RELNAME
END PRE_SQL,
'1' STATUS,
T2.SCHEMANAME LANDING_SCHEMA
FROM PG_CLASS T
LEFT JOIN PG_PARTITION T1 ON T.OID = T1.PARENTID AND T.RELNAME <> T1.RELNAME
LEFT JOIN PG_TABLES T2 ON T.RELNAME = T2.TABLENAME AND T2.TABLENAME <> 't18_sync'
AND T2.SCHEMANAME = 'aml' AND T2.TABLEOWNER = 'aml'
WHERE T2.TABLENAME IS NOT NULL
ORDER BY T.RELNAME, T1.RELNAME;
```
初始化完成后，再次执行同步工具
```sh
java -jar DataSync-x.x.x.jar -o 写出数据源名 -i 写入数据源名
```
...
6. 也可以自己自定义在T18_SYNC添加数据，指定WHERE条件，查询字段数，落地到什么表
```sql
INSERT INTO T18_SYNC (sync_sql, status, landing_tablename)
values ('SELECT USERNAME,REALNAME FROM T00_USER WHERE USERNAME = ''admin'' ', '1', 'T00_USER_BAK');
```
```sql
SELECT * FROM T18_SYNC;
```
| sync_sql | pre_sql | begin_date | end_date | status | error_msg | sync_data_size | landing_tablename | landing_schema |
| ------------------------------------------------------------ | ------- | ---------- | -------- | ------ | --------- | -------------- | :---------------- | -------------- |
| SELECT USERNAME,REALNAME FROM T00_USER WHERE USERNAME = 'admin' | | | | 1 | | | T00_USER_BAK | |
最后再执行工具
```sh
java -jar DataSync-x.x.x.jar -o 写出数据源名 -i 写入数据源名
```
7. 在后台批处理调度任务中执行，需要将DataSync-x.x.x.jar、config.properties、lib驱动jar包目录，放入项目中的resources目录
```java
public class ExecSyncTask {
private static final String JAR = "DataSync-1.1.8.jar";
private static final Logger logger = LoggerFactory.getLogger(ExecDubboTask.class);
public static boolean exec(T18_job_instanceDTO t18_job_instance) throws Exception {
// 获取路径
String path = ExecSyncTask.class.getResource(File.separator + JAR).getPath();
// 拟定执行命令
String[] command = {"java", "-jar", path, "-o", "db2", "-i", "openGauss", "-t", t18_job_instance.getJob_model_id()};
ProcessBuilder processBuilder = new ProcessBuilder(command);
processBuilder.inheritIO();
// 开始执行
Process process = processBuilder.start();
// 获取输出
BufferedReader processInput = new BufferedReader(new InputStreamReader(process.getInputStream()));
// 获取错误输出
BufferedReader processError = new BufferedReader(new InputStreamReader(process.getErrorStream()));
String line;
while ((line = processInput.readLine()) != null){
// 打印日志
logger.info(line);
}
while ((line = processError.readLine()) != null){
// 打印错误日志
logger.error(line);
}
// 获取执行状态 0 成功 1 失败
int status = process.waitFor();
if(status == 0){
return true;
}else{
return false;
}
}
}
```
配置相应的JOB任务，最后执行
##### 表结构
1. 从其他数据库生成表结构，生成完毕后获得create_时间戳.sql文件，将SQL语句在新库执行。
```sh
java -jar DataSync-x.x.x.jar -m create -o 数据源
```
...
2. 从GaussDB生成表结构，生成完毕后获得create_时间戳.sql文件，将SQL语句在新库执行。
```sh
java -jar DataSync-x.x.x.jar -m create -o 数据源
```
...
3. 从GaussDB生成部分表结构，生成完毕后获得create_时间戳.sql文件
```sh
java -jar DataSync-x.x.x.jar -m create -o 数据源 -t 表名(例如:T00_USER或者T00_U%模糊匹配)
```
##### 数据倾斜
1. 统计GaussDB(DWS)数据倾斜问题，该功能对openGauss及其他数据库无效！
```sh
java -jar DataSync-x.x.x.jar -m skewness -o 数据源
```
...
执行完毕后，会生成两种日志文件，一个是执行过程中的全部日志，包含全部表的分析数据。一个是异常日志，数据量相差5%以上即可视为倾斜，如果相差10%以上就必须要调整分布列。
2. 统计GaussDB(DWS)指定表数据倾斜问题
```sh
java -jar DataSync-x.x.x.jar -m skewness -o 数据源 -t 表名(例如:T00_USER或者T00_U%模糊匹配)
```
##### 表结构对比
1. 对比Oracle和GaussDB(DWS)表结构差异
```sh
java -jar DataSync-x.x.x.jar -m compare -d 多数据源名,以逗号分隔 -t 表名(例如:T00_USER或者T00_U%模糊匹配)
```
#### 常见错误
1. 找不到xxx.xxx.Driver驱动包，如右图
- 解决办法：将相应JDBC驱动包放入DataSync-x.x.x.jar同级目录中的lib文件夹中。
#### 需求缺陷
1. 发现缺陷可以直接发邮件进行反馈
2. 有新的需求or好的idea也可以直接与我联系沟通
#### 版本说明
v1.2.0 新增表结构对比功能
v1.1.10 通过代码安全检查，修复存在的隐患
v1.1.9 新增导出字段全部NOT NULL，修复表结构导出默认值未导出问题。
v1.1.8 修复多线程，线程不安全问题; 新增会话超时时间设置
v1.1.7 新增数据库配置Schema主要应用于Gauss数据库有多模式选择
v1.1.6 修复高版本Oracle驱动获取用户模式异常问题
v1.1.5 修复数据同步时若含有(\\.)导致数据同步失败问题