实用工具之DataXWeb数据同步

1.前言

DataX实用json来描述数据同步任务，比较麻烦，所以有人开发了基于DataX的Web可视化界面，就是DataX-Web。这里有一个问题，就是DataX-Web本身没有starrocks执行器，但是它是调用的独立安装的DataX，所以如果你自己安装了starrocks的writer，那么就可以在构建执行器的时候，修改生成的json代码，这样就实现了starrocks的读取写入。

2.安装

安装其实挺简单的，先安装datax源码，也是下载解压就好了，然后指定环境变量，或者是修改安装目录下的 modules/datax-executor/bin/env.properties 文件，找到 PYTHON_PATH 指向 datax.py 脚本的位置，详情请查看问题说明。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

## 编译安装
## 直接从Git上面获得源代码，在项目的根目录下执行如下命令
mvn clean install 
## 执行成功后将会在工程的build目录下生成安装包
build/datax-web-{VERSION}.tar.gz

## 为了顺利启动，最好是这是环境变量，
vi /etc/profile
## 设置DATAX_HOME目录
export DATAX_HOME=/usr/local/datax/bin/

## 在选定的安装目录，解压安装包
tar -zxvf datax-web-{VERSION}.tar.gz
## 交互式安装
./bin/install.sh
## 一键启动所有服务
./bin/start-all.sh
## 一键取消所有服务
./bin/stop-all.sh

3.使用

打开 ip:9527端口，进行任务创建。http://localhost:9527/index.html, 默认管理员用户名：admin 密码：123456

(1) 项目管理->创建项目，指定相应的项目名称

(2) 数据源管理->创建数据源，实用jdbc连接相应的数据库

(3) 任务管理->DataX任务模版，创建一个DataX任务模版

(4) 任务管理->任务构建，构建任务

(4) 任务管理->任务管理，执行任务

4.增量更新配置

其实我有点蒙，我有一张表叫 realtime ，其中的一个字段叫 RecordTime，是从2012年开始的 yyyy-MM-dd HH:mm:ss 字段格式，我想着用这个字段进行增量配置。主要配置,辅助参数：时间自增；增量开始时间：选择一个时间；增量时间字段：-DlastTime=’%s’ -DcurrentTime=’%s’；增量时间格式：时间戳。

修改生成的json文件，将table去掉，加上querySql语句,这里之所以使用了 1970-01-01 08:00:00，是使用的北京时间的unix时间戳的开始时间,DATEADD第一个参数 S 指的是秒数。

1
2
3
4
5
6

"connection": [
    {
    "querySql": [
        "select * from realtime where RecordTime >= DATEADD(S,convert(int,${lastTime}),'1970-01-01 08:00:00') and RecordTime < DATEADD(S,convert(int,${currentTime}),'1970-01-01 08:00:00') "
    ]
}

还有就是要使用 convert 函数，将 ${lastTime} 参数进行转换，否在生成的时间戳就变成了字符串，提示：参数数据类型 varchar 对于 dateadd 函数的参数 2 无效

1

select * from realtime where RecordTime >= DATEADD(S,'1646903822','1970-01-01 08:00:00') and RecordTime < DATEADD(S,'1647512519','1970-01-01 08:00:00') 

5.解决增量更新的过程

[尝试]
(1) 仓库里面的说明，需要配置querSql，在增量字段里面配置了：-DlastTime=’%s’ -DcurrentTime=’%s’，但是这样配置之后，日志里面出现了：您的配置凌乱了. 因为datax不能同时既配置table又配置querySql.请检查您的配置并作出修改.

1
2
3
4
5
6
7
8
9
10
11
12
13

"connection": [
    {
        "table": [
            "realtime"
        ],
        "jdbcUrl": [
            "jdbc:sqlserver://xxx:xxx;DatabaseName=phems"
        ],
        "querySql": [
            "select * from realtime where RecordTime >= FROM_UNIXTIME(${lastTime}) and RecordTime < FROM_UNIXTIME(${currentTime})"
        ]
    }
]

(2) 后来我把table这个字段删掉，修改了connection，只留下了querySql，

1
2
3
4

"querySql": [
    "select * from realtime where RecordTime >= FROM_UNIXTIME(${lastTime}) and RecordTime < FROM_UNIXTIME(${currentTime})"
]

出现了错误：您的配置有误. 由于您读取数据库表采用了querySql的方式, 所以您不需要再配置 column. 如果您不想看到这条提醒，请移除您源头表中配置中的 column.

(3) 然后我回退到table，删除了querySql,我把增量字段改成了 RecordTime

结果出现了错误: 找不到或无法加载主类 RecordTime

(4) 我再次尝试将增量字段设置为：-DlastTime=’%s’ -DcurrentTime=’%s’，配置了querySql,将 “column” 配置为 [ “*” ],结果出现了：执行的SQL为: select * from realtime where RecordTime >= FROM_UNIXTIME(‘2010-01-01%00:00:00’) and RecordTime < FROM_UNIXTIME(‘2022-01-10%18:13:45’) 具体错误信息为：com.microsoft.sqlserver.jdbc.SQLServerException: ‘FROM_UNIXTIME’ 不是可以识别的 内置函数名称。

这个可能是因为我的数据库是SQL Server的原因。

(5) 删掉了FROM_UNIXTIME函数，还是出现了问题：执行的SQL为: select * from realtime where RecordTime >= ‘2010-01-01%00:00:00’ and RecordTime < ‘2022-01-10%18:21:28’ 具体错误信息为：com.microsoft.sqlserver.jdbc.SQLServerException: 从字符串转换日期和/或时间时，转换失败,也就是说在生成的增量SQL语句时，将日期格式中间加了一个 ‘%’， 这就无法执行SQL 语句了。

[尝试]
有人说把增量语句改为：-DlastTime=’yyyy-MM-dd ‘ -DcurrentTime=’yyyy-MM-dd’，就可以了,我尝试之后，还是不行。

(6) 阅读源码 src/main/java/com/wugui/datax/executor/service/command/BuildCommand.java。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

if (IncrementTypeEnum.TIME.getCode() == incrementType) {
    if (doc.length() > 0) doc.append(SPLIT_SPACE);
    String replaceParamType = tgParam.getReplaceParamType();

    if (StringUtils.isBlank(replaceParamType) || replaceParamType.equals("Timestamp")) {
        long startTime = tgParam.getStartTime().getTime() / 1000;
        long endTime = tgParam.getTriggerTime().getTime() / 1000;
        doc.append(PARAMS_CM).append(TRANSFORM_QUOTES).append(String.format(replaceParam, startTime, endTime));
    } else {
        SimpleDateFormat sdf = new SimpleDateFormat(replaceParamType);
        String endTime = sdf.format(tgParam.getTriggerTime()).replaceAll(SPLIT_SPACE, PERCENT);
        String startTime = sdf.format(tgParam.getStartTime()).replaceAll(SPLIT_SPACE, PERCENT);
        doc.append(PARAMS_CM).append(TRANSFORM_QUOTES).append(String.format(replaceParam, startTime, endTime));
    }
    //buildPartitionCM(doc, partitionStr);
    doc.append(TRANSFORM_QUOTES);

} else if (IncrementTypeEnum.ID.getCode() == incrementType) {
    long startId = tgParam.getStartId();
    long endId = tgParam.getEndId();
    if (doc.length() > 0) doc.append(SPLIT_SPACE);
    doc.append(PARAMS_CM).append(TRANSFORM_QUOTES).append(String.format(replaceParam, startId, endId));
    doc.append(TRANSFORM_QUOTES);
}

时间戳的方式，我想起来了，可以传递时间戳的方式，但是在时间字符串上进行下功夫，将 “2022-01-02 13:00:00” 转换为时间戳，进行比较。因为Datax-Web,会最终将querySql中的变量，进行替换。

1
2
3
4
5
6
7
8
9
10
11

--- 增量开始时间：2022-03-10 17:17:02，增量时间字段：-DlastTime='%s' -DcurrentTime='%s'增量时间格式：时间戳
--- json中的 querySql
select * from realtime where RecordTime >= (${lastTime}) and RecordTime < (${currentTime})
--- 最后转换的结果就是
select * from realtime where RecordTime >= ('1646903822') and RecordTime < ('1647509572')

--- 方法一：如果RecordTime格式为:yyyy-MM-dd HH:mm:ss,那么可以将该时间字段，使用sql语句转换为时间戳再进行比较
SELECT * from realtime r WHERE DATEDIFF(S,'1970-01-01 08:00:00', RecordTime) > 1646903822

--- 方法二：将生成的时间戳，使用sql语句，对时间戳进行时间格式化，转换为 yyyy-MM-dd HH:mm:ss 在使用 >= 进行比较,这个方法效率要高很多
select * from realtime where RecordTime >= DATEADD(S,${lastTime},'1970-01-01 08:00:00') and RecordTime < DATEADD(S,${currentTime},'1970-01-01 08:00:00') 

/usr/bin/python: can’t find ‘main‘ module in ‘’

安装datax-web，增加执行器之后，执行过程中，出现了错误：

1
2
3
4
5
6
7
8
9

## 编辑配置文件
vi /usr/local/datax-web-2.1.2/modules/datax-executor/bin/env.properties
## 修改,脚本执行器
PYTHON_PATH=/usr/local/datax/bin/datax.py

## 或者修改
vi /etc/profile
## 设置DATAX_HOME目录
export DATAX_HOME=/usr/local/datax/bin/

当 IDENTITY_INSERT 设置为 OFF 时，不能为表中的标识列插入显式值

出现这个错误，其实就是因为主键自增id不能被明显的设置，比如，如果你将id设置为自增，在进行同步的时候，就不能插入这个自增的字段。

【尝试方案】
（1）最简单的方式，就是不插入 id，这种自动增长的列。
（1）设置 IDENTITY_INSERT
当我尝试设置这个值之后，发现总是无效的。

1

SET IDENTITY_INSERT tablename ON

注意
1.每一次连接会话中的任一时刻，只能对一个表设置 IDENTITY_INSERT ON，且设置只对当前会话有效；
2.在对标识列执行插入操作进，一定要列出此标识列（当然，同时也就需要列出相关的其他列了）。

（2）尝试使用 preSql语句
在执行插入之前，先执行 set IDENTITY_INSERT语句。这个参数在 parameter 下面。

1
2
3

"preSql": [
    " set IDENTITY_INSERT dbo.realtime on"
],

【解决方案】
我升级了 datax 版本，2023年09月之后，这个问题就解决了，然后 尝试使用 session 配置，最后终于可以同步数据了。

1
2
3
4
5
6

"writer": {
    "name": "sqlserverwriter",
    "parameter": {
        "session": ["SET IDENTITY_INSERT tableName ON"]
    }
}

“exception”:”将 nvarchar 转换为数据类型 numeric 时出现算术溢出错误。

这个问题比较奇怪了，明明Reader和Writer的数据库的表结构都是一样的，最后在执行的时候，还是出现了这个问题，两个数据库都是SqlServer