springboot整合datax的使用

datax资源下载

datax 项目集成

1、首先将下载的datax文件中的Datax集成到我们spring boot + cloud中
先将代码文件夹 copy到 项目中 然后将其设置为 module

再修改 相应POM文件

• 1、查看 项目主文件pom
  

• 2、将 datax主文件pom进行修改
  

• 3、修改主文件pom，增加两项
  
  集成完成

datax的使用

这里介绍的是 mysql数据迁移，将一个表的10W个数据转移到另一张表

• 1、首先将 下载的datax文件 COPY到要使用的 module的rescource中
  

• 2、在开始使用 datax的地方配置一下
  此时默认调用的是 entry 方法，entry200这个方法是我自己写的
  

• 3、 Datax json的介绍

{
    "job": {
        "content": [{
            "reader": { 
                "name": "mysqlreader",  
                "parameter": {
                    "column": [ // 源表的字段
                        "doc_id",
                        "title",
                        "file_path",
                        "approval_id",
                        "page_count",
                        "version"
                    ],
                    "connection": [{
                        "jdbcUrl": [ // 源表地址
                            "jdbc:mysql://192.168.81.1:3306/bootdo?useUnicode=true&characterEncoding=utf8"
                        ],
                        "table": [ // 源表的 表名
                            "es_approval_doc"
                        ]
                    }],
                    "password": "123456", // 源表用户名 密码
                    "username": "root",
                    "where": "version > FROM_UNIXTIME(${start_time}) and version"  // 可以不加，加的话就是对表数据进行筛选，不能使用limit
                }
            },
            "writer": {
                "name": "mysqlwriter",
                "parameter": {
                    "column": [ // 目的表字段，要和源表一致
                        "doc_id",
                        "title",
                        "file_path",
                        "approval_id",
                        "page_count",
                        "version"
                    ],
                    "writeMode": "update",
                    "connection": [{
                        "jdbcUrl": "jdbc:mysql://192.168.81.1:3306/bootdo?useUnicode=true&characterEncoding=utf8",
                        "table": [
                            "es_approval_doc_copy"
                        ]
                    }],
                    "password": "123456",
                    "username": "root"
                }
            }
        }],
        "setting": {
            "speed": {
                "channel": "1"
            }
        }
    }
}
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

写好这个后，根据2调用的方法
可以直接将 源表的数据 copy到 目的表

• 4、因为 where 中的 ${参数} 我不会传，并且 limit 不允许在where的条件中，所以我重写了一个 entryCopy200 方法以代码的形式将sql语句传引擎，和 在json中获取数据一致

Configuration configuration = engineEntry(args);
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String strTime = sdf.format(time);
//        strTime = "2022-11-01 08:17:21";
        // set param
        JSONObject jobObj = JSONObject.parseObject(configuration.get("job", JSONObject.class).toString());
        if (jobObj != null) {
            String querySql = "select `id`,`username`,`age`,`create_time` from `user` where `create_time` > "+"'" + strTime + "'"+" limit 200";
            ArrayList<String> sqlList = new ArrayList<>();
            sqlList.add(querySql);
            JSONArray cintentArr = jobObj.getJSONArray("content");
            cintentArr.getJSONObject(0).getJSONObject("reader").getJSONObject("parameter").getJSONArray("connection").getJSONObject(0).put("querySql",sqlList);

            jobObj.put("content", cintentArr);
            configuration.set("job", jobObj);
        }
        ConfigurationValidate.doValidate(configuration);
        Engine engine = new Engine();
        engine.start(configuration);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

我们的json文件的数据都会被读取到 configuration 这个对象中，我们可以在读取数据之后，再以代码形式 set进configuration 。

类似querySql 的用法 可以参考datax.json详解
这个也不错

问题

1、

从源码出拉取需要的代码，将相应模块copy到 需要的位置，编译安装打包后，将

打包的文件夹下的reader copy到 用 datax的地方

即可解决。