实用工具之DataX数据同步

标签： 无 分类： 未分类 创建时间：2021-12-23 01:53:18 更新时间：2024-11-23 10:24:25

前言

本来还在考虑要不要自己写一个工具，将老系统中的SQL Server的数据迁移出来，因为服务器特别的老，数据也比较多，想写一个同步工具，后来今天还看到了数据自动同步的工具，比如：DataX、Tis

参考文章:
1.基于TIS快速实现MySQL到StarRocks的实时数据同步方案
2.各种开源数据库同步工具汇总
3.dataX是阿里开源的离线数据库同步工具的使用
4.阿里云开源离线同步工具DataX3.0介绍
5.alibaba /DataX
6.几种主流数据同步/ETL工具的比较 TurboDX、Oracle Goldengate、Kettle、DataX、Informatica
7.DataX在有赞大数据平台的实践
8.DataX管理系统datax-web安装部署记录

1.下载

到官网下载安装包

2.安装

只需要解压就可以进行安装了，但是需要的依赖，包括java、maven和python，python最好是python2,如果不是的话，还要修改datax的执行脚本，因为 datax.py 使用的是python2编写的

3.编写json

这里我测试的是从sqlserver导入数据到starrocks中。到starrocks官网下载 writer 插件，放到datax的插件文件夹中，我这里是 /usr/local/datax/plugin/writer 目录。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
{
    "job": {
        "setting": {
            "speed": {
                #数据分片，分片数据可同时进行同步
                "channel": 128 
            }
        },
        "content": [{
             #SQL Server配置
            "reader": {
                "name": "sqlserverreader",
                "parameter": {
                    #数据库用户名
                    "username": "${srcUserName}",
                    #数据库密码
                    "password": "${srcPassword}",
                     #需要迁移的列名，* 表示全部列
                    "column": ["*"]
                    "connection": [{
                        #需要迁移的表名
                        "table": ["${tableName}"],
                        #数据库 jdbc 连接，比如 jdbc:sqlserver://192.168.1.99:1433;databasename=databasename
                        "jdbcUrl": ["${srcUrl}"],
                        # 当用户配置querySql时，MysqlReader直接忽略table、column、where条件的配置，querySql优先级大于table、column、where选项
                        "querySql": [
                            ""
                        ]
                    }]
                }
        },
        #TiDB配置
        "writer": {
                "name": "mysqlwriter",
                "parameter": {
                    #数据库用户名
                    "username": "${desUserName}",
                    #数据库密码
                    "password": "${desPassword}",
                    #使用 Replace 语法
                    "writeMode": "replace",
                    #目标表列名，* 表示全部列
                    "column": ["*"],
                    "connection": [{
                        #数据库 jdbc 连接，比如 jdbc:mysql://192.168.1.95:9030/databasename
                        "jdbcUrl": "${desUrl}",
                        #目标表名
                        "table": ["${tableName}"]
                    }],
                    #本次迁移开始前执行的sql-作数据迁移日志使用
                    "preSql": [],
                    #本次迁移完成后执行的 sql- 作数据迁移日志使用
                    "postSql": []
                }
            }
        }]
    }
}

从sqlserver 导出数据导starrocks示例:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
{
    "job": {
        "setting": {
            "speed": {
                "channel": 128 
            }
        },
        "content": [{
            "reader": {
                "name": "sqlserverreader",
                "parameter": {
                    "username": "xxx",
                    "password": "xxx",
                    "column": [
                        "[RecordTime]",
                    ],
                    "connection": [{
                        "table": ["tablename"],
                        "jdbcUrl": ["jdbc:sqlserver://ip:port;databasename=databasename"]
                    }]
                }
            },
            "writer": {
                "name": "starrockswriter",
                "parameter": {
                    "username": "xxx",
                    "password": "xxx",
                    "database": "databasename",
                    "table": "tablename",
                    "column": [
                        "`RecordTime`",
                    ],
                    "preSql": [],
                    "postSql": [], 
                    "jdbcUrl": "jdbc:mysql://192.168.1.95:9030/",
                    "loadUrl": ["192.168.1.95:8030"],
                    "loadProps": {}
                }
            }
        }]
    }
}
参考文章:
1.5.4 SqlServer 到 TiDB（DATAX）
2.DataX doriswriter
3.StarRocksWriter StarRocks编写的可以在DataX使用的插件

4.执行数据迁移

1
2
3
4
5
6
python /usr/local/datax/bin/datax.py /usr/local/datax/job/full_starrocks.json

## 后台执行
nohup python /usr/local/datax/bin/datax.py /usr/local/datax/job/lsmfhx_realtime.json > lsmfhx_realtime.log 2>&1 &
## 查看
ps aux|grep python
参考文章:
1.datax入门到实战及面试 这里使用的是源码编译的方式安装的Datax，还有关于面试的一些问题
2.datax的安装与使用
3.DataX的Clickhouse读写插件
4.使用Datax同步mysql,oracle,sqlserver数据到Hbase 这里提到了一个datax-web，应该是一个可视化的配置页面，可以修改json数据内容

5.增量同步

因为每一次使用python进行同步的时候，所有的数据都会重新进行同步一遍，明显的就是说，使用Ctrl+C结束进程，然后再重新进行同步的时候，所有的记录又都查询了一遍。增量同步的意思，就是重复的记录就不要进行重复同步了，这样也不用担心数据会重复存储了。

参考文章:
1.DataX 数据全量，增量同步方案 这里提供了一个shell脚本，将获取到的时间写入csv文件中，然后动态修改json文件，达到增量更新的目的
2.datax调研及增量更新的思路 这里提到了一个 DataX-Migration 的工具。
3.Jawf/DataX-Migration
4.使用 DataX 增量同步数据
5.DataX （原理， 增量全量同步数据实践） 这里是通过在json中，通过querySQL实现的

问题

1.plugin.json 不存在

配置信息错误，您提供的配置文件[/usr/local/datax/plugin/reader/._drdsreader/plugin.json]不存在. 请检查您的配置文件

[解决方法]

1
2
3
4
5
6
7
8
9
10
## 进入
cd plugin/reader

## 列出全部文件
ls -al

## 删除以 ._ 开头的文件
rm -rf ._*

## 同样的删除writer目录下的全部的以 ._ 开头的文件
参考文章:
1.plugin.json文件不存在，请检查您的配置文件
2.下载下来的二进制包启动报错 datax/plugin/reader/._drdsreader/plugin.json

2.postgresqlwrite字段名大写问题

如果postgresql数据库中的字段名是大写的情况下，那么在 “column”: [] 的字段配置中，就应该这么写 “column”: [“"RecordTime"“]，而不是 “column”: [“RecordTime"]。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
"writer": {
    "name": "postgresqlwriter",
    "parameter": {
        "column": [
            "\"RecordTime\"",
        ],
        "connection": [
            {
                "jdbcUrl": "jdbc:postgresql://192.168.1.98:5432/hlgy",
                "table": ["realtime"]
            }
        ],
        "password": "xxx",
        "postSql": [],
        "preSql": [],
        "username": "xxx"
    }
}

mysql的语句应该是 “column”: [“RecordTime“]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
"reader": {
    "name": "mysqlreader",
    "parameter": {
        "column": [
            "`RecordTime`",   
        ],
        "connection": [
            {
                "jdbcUrl": ["jdbc:mysql://192.168.1.95:9030/hlgy"],
                "table": ["realtime"]
            }
        ],
        "password": "xxx",
        "username": "xxx",
        "where": ""
    }
}

3.在有总bps限速条件下，单个channel的bps值不能为空，也不能为非正数

修改 datax/conf/core.json 文件，修改core -> transport -> channel -> speed -> “byte”: 2000000，将单个channel的大小改为2MB即可。

参考文章:
【1】.DataX报错解决办法 - 在有总bps限速条件下，单个channel的bps值不能为空，也不能为非正数 修改文件datax/conf/core.json
**本文已结束，感谢您的观看**
鼓励一下

往期推荐

文章目录
  1. 1. 前言
  2. 2. 1.下载
  3. 3. 2.安装
  4. 4. 3.编写json
  5. 5. 4.执行数据迁移
  6. 6. 5.增量同步
  7. 7. 问题
    1. 7.1. 1.plugin.json 不存在
    2. 7.2. 2.postgresqlwrite字段名大写问题
    3. 7.3. 3.在有总bps限速条件下，单个channel的bps值不能为空，也不能为非正数
小额赞助
本人提供免费与付费咨询服务，感谢您的支持！赞助请发邮件通知，方便公布您的善意！
**光 3.01 元
Sun 3.00 元
bibichuan 3.00 元
捐赠者名单 合计：9.01 元
微信公众号
广告位
诚心邀请广大金主爸爸洽谈合作
每日一省
isNaN 和 Number.isNaN 函数的区别？

1.函数 isNaN 接收参数后，会尝试将这个参数转换为数值，任何不能被转换为数值的的值都会返回 true，因此非数字值传入也会返回 true ，会影响 NaN 的判断。

2.函数 Number.isNaN 会首先判断传入参数是否为数字，如果是数字再继续判断是否为 NaN ，不会进行数据类型的转换，这种方法对于 NaN 的判断更为准确。

2021-04-07 链接地址
每日二省
为什么0.1+0.2 ! == 0.3，如何让其相等?

一个直接的解决方法就是设置一个误差范围，通常称为“机器精度”。对JavaScript来说，这个值通常为2-52，在ES6中，提供了Number.EPSILON属性，而它的值就是2-52，只要判断0.1+0.2-0.3是否小于Number.EPSILON，如果小于，就可以判断为0.1+0.2 ===0.3。

2021-04-07 链接地址
每日三省
== 操作符的强制类型转换规则？

1.首先会判断两者类型是否**相同，**相同的话就比较两者的大小。

2.类型不相同的话，就会进行类型转换。

3.会先判断是否在对比 null 和 undefined，是的话就会返回 true。

4.判断两者类型是否为 string 和 number，是的话就会将字符串转换为 number。

5.判断其中一方是否为 boolean，是的话就会把 boolean 转为 number 再进行判断。

6.判断其中一方是否为 object 且另一方为 string、number 或者 symbol，是的话就会把 object 转为原始类型再进行判断。

2021-04-07 链接地址
每日英语
Happiness is time precipitation, smile is the lonely sad.
幸福是年华的沉淀，微笑是寂寞的悲伤。
2016 ——— 美文