【DataX】Java中集成DataX开发

简介: 利用java集成datax开发,非java调用python方式。

本文链接

步骤

先说总体步骤:

  1. 下载源码,并编译到本地maven仓库[上传私服(可选)];
  2. pom文件依赖datax-core和需要的readerwriter
  3. 环境变量设置datax.home(或者利用System#setProperty(String))和一些需要替换脚本中的变量:脚本中${}占位符的变量将被系统变量替换。
  4. 将datax.tar.gz中解压出来的confplugin等文件放到datax.home目录中。
  5. 构造参数数组:{"-job", "xxx.json", "-mode", "standalone", "-jobid", "-1"}
  6. 调用Engin#main(String[])或者Engine#entry(String[])

引言

目前官方的使用指南里都是利用python来调用dataX执行任务。而且现有的博客基本上也是利用java来调用python命令Runtime.getRuntime().exec()来执行。
个人感觉,dataX未提供java集成开发的方法,应该是定位生产系统,运维需要吧?!
我们的业务场景:执行完dataX的job之后,还有一定的业务逻辑,所以希望在java应用里调用dataX执行完job之后,再执行后续逻辑。

DataX分析

笔者简单的看了一下午的DataX的逻辑,完全以使用者的视角分析DataX,必然不能完全了解DataX的整个执行过程。
本文仅分析如果能够在java代码里集成DataX进行开发。

集成准备

DataX没有将代码上传到maven服务器上,所以需要自己先pull代码到本地,编译,才能在集成开发的使用通过pom引用。有条件的可以上传到自己的私服上。
代码地址

代码依赖

通过pom文件加入datax-core

<dependency>
    <groupId>com.alibaba.datax</groupId>
    <artifactId>datax-core</artifactId>
    <version>0.0.1-SNAPSHOT</version>
</dependency>

如果需要对应的readerwriter的话,加入到pom文件中,比如需要streamreader和streamwriter:

<dependency>
    <groupId>com.alibaba.datax</groupId>
    <artifactId>streamreader</artifactId>
    <version>0.0.1-SNAPSHOT</version>
</dependency>
<dependency>
    <groupId>com.alibaba.datax</groupId>
    <artifactId>streamwriter</artifactId>
    <version>0.0.1-SNAPSHOT</version>
</dependency>

要依赖datax一定要保证有对应的源码或者编译到本机的maven repository或者在对应的私服上有上传相应的编译版本,不然pom文件是找不到依赖的。

为了集成开发,可能需要一口气引用所有的reader和writer,目前所知,就得一个一个写,如果大家有好办法,麻烦告知!

准备相应的文件

com.alibaba.datax.core.util.container.CoreConstant中可以看到,datax.home很重要,很多文件的读取都是在datax.home里面获取的。就如我们在安装版的datax中可以看到里面一些目录一样

$ ll
total 4
drwxr-xr-x 2 mcbadm mcb   56 Sep 20 18:28 bin
drwxr-xr-x 2 mcbadm mcb   65 Sep 20 18:28 conf
drwxr-xr-x 2 mcbadm mcb   21 Sep 20 18:28 job
drwxr-xr-x 2 mcbadm mcb 4096 Sep 20 18:28 lib
drwxr-xr-x 4 mcbadm mcb   32 Sep 20 18:28 plugin
drwxr-xr-x 2 mcbadm mcb   22 Sep 20 18:28 script
drwxr-xr-x 2 mcbadm mcb   23 Sep 20 18:28 tmp

目前所知的,Engine#entry在解析配置的时候会读取conf目录下的文件,还有对应plugin/reader/xxxreader、plugin/writer/xxxwriter的plugin.json文件:

{
    "name": "streamreader",
    "class": "com.alibaba.datax.plugin.reader.streamreader.StreamReader",
    "description": {
        "useScene": "only for developer test.",
        "mechanism": "use datax framework to transport data from stream.",
        "warn": "Never use it in your real job."
    },
    "developer": "alibaba"
}

编写代码

编写job代码:

{
  "job": {
    "content": [
      {
        "reader": {
          "name": "streamreader",
          "parameter": {
            "sliceRecordCount": 1,
            "column": [
              {
                "type": "long",
                "value": "10"
              },
              {
                "type": "string",
                "value": "hello,你好,世界-DataX"
              }
            ]
          }
        },
        "writer": {
          "name": "streamwriter",
          "parameter": {
            "encoding": "UTF-8",
            "print": true
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "channel": 1
       }
    }
  }
}

写个测试类吧:

import com.alibaba.datax.core.Engine;

public class EngineTest {
    
    public static void main(String[] args) {
        System.setProperty("datax.home", getCurrentClasspath());
        String[] datxArgs = {"-job", getCurrentClasspath() + "/job/stream2stream.json", "-mode", "standalone", "-jobid", "-1"};
        try {
            Engine.entry(datxArgs);
        } catch (Throwable e) {
            e.printStackTrace();
        }
    }
    
    public static String getCurrentClasspath() {
        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
        String currentClasspath = classLoader.getResource("").getPath();
        // 当前操作系统
        String osName = System.getProperty("os.name");
        if (osName.startsWith("Windows")) {
            // 删除path中最前面的/
            currentClasspath = currentClasspath.substring(1);
        }
        return currentClasspath;
    }
}

datax在解析完配置后,会将core.json,job.json,plugin.json合并在一起:

{
    "common": {
        "column": {
            "dateFormat": "yyyy-MM-dd", 
            "datetimeFormat": "yyyy-MM-dd HH:mm:ss", 
            "encoding": "utf-8", 
            "extraFormats": [
                "yyyyMMdd"
            ], 
            "timeFormat": "HH:mm:ss", 
            "timeZone": "GMT+8"
        }
    }, 
    "core": {
        "container": {
            "job": {
                "id": -1, 
                "reportInterval": 10000
            }, 
            "taskGroup": {
                "channel": 5
            }, 
            "trace": {
                "enable": "false"
            }
        }, 
        "dataXServer": {
            "address": "http://localhost:7001/api", 
            "reportDataxLog": false, 
            "reportPerfLog": false, 
            "timeout": 10000
        }, 
        "statistics": {
            "collector": {
                "plugin": {
                    "maxDirtyNumber": 10, 
                    "taskClass": "com.alibaba.datax.core.statistics.plugin.task.StdoutPluginCollector"
                }
            }
        }, 
        "transport": {
            "channel": {
                "byteCapacity": 67108864, 
                "capacity": 512, 
                "class": "com.alibaba.datax.core.transport.channel.memory.MemoryChannel", 
                "flowControlInterval": 20, 
                "speed": {
                    "byte": -1, 
                    "record": -1
                }
            }, 
            "exchanger": {
                "bufferSize": 32, 
                "class": "com.alibaba.datax.core.plugin.BufferedRecordExchanger"
            }
        }
    }, 
    "entry": {
        "jvm": "-Xms1G -Xmx1G"
    }, 
    "job": {
        "content": [
            {
                "reader": {
                    "name": "streamreader", 
                    "parameter": {
                        "column": [
                            {
                                "type": "long", 
                                "value": "10"
                            }, 
                            {
                                "type": "string", 
                                "value": "hello,你好,世界-DataX"
                            }
                        ], 
                        "sliceRecordCount": 1
                    }
                }, 
                "writer": {
                    "name": "streamwriter", 
                    "parameter": {
                        "encoding": "UTF-8", 
                        "print": true
                    }
                }
            }
        ], 
        "setting": {
            "speed": {
                "channel": 1
            }
        }
    }, 
    "plugin": {
        "reader": {
            "streamreader": {
                "class": "com.alibaba.datax.plugin.reader.streamreader.StreamReader", 
                "description": {
                    "mechanism": "use datax framework to transport data from stream.", 
                    "useScene": "only for developer test.", 
                    "warn": "Never use it in your real job."
                }, 
                "developer": "alibaba", 
                "name": "streamreader", 
                "path": "D:/workspace/datax-test/target/test-classes/\\plugin\\reader\\streamreader"
            }
        }, 
        "writer": {
            "streamwriter": {
                "class": "com.alibaba.datax.plugin.writer.streamwriter.StreamWriter", 
                "description": {
                    "mechanism": "use datax framework to transport data to stream.", 
                    "useScene": "only for developer test.", 
                    "warn": "Never use it in your real job."
                }, 
                "developer": "alibaba", 
                "name": "streamwriter", 
                "path": "D:/workspace/datax-test/target/test-classes/\\plugin\\writer\\streamwriter"
            }
        }
    }
}

说说插件原理

每个reader和writer都有自己的plugin.json文件,里面最重要的就是class配置了,这个类的全路径配置用于classloader将其加载进来并通过反射将其实例化。加载代码可看com.alibaba.datax.core.util.container.LoadUtil
所以我们在集成的时候,plugin目录下面不需要有jar包了,只需要放json文件就行,因为我们通过pom文件依赖了对应的reader和writer,说白了,就是classpath下面有对应的reader和writer即可。

结束语

文章有点长,记录了一个下午的研究结果,应该有很多不完善的地方,希望可以和大家多交流。如果觉得有帮助,可以点个赞。

目录
相关文章
|
18天前
|
前端开发 JavaScript Java
基于Java+Springboot+Vue开发的服装商城管理系统
基于Java+Springboot+Vue开发的服装商城管理系统(前后端分离),这是一项为大学生课程设计作业而开发的项目。该系统旨在帮助大学生学习并掌握Java编程技能,同时锻炼他们的项目设计与开发能力。通过学习基于Java的服装商城管理系统项目,大学生可以在实践中学习和提升自己的能力,为以后的职业发展打下坚实基础。
51 2
基于Java+Springboot+Vue开发的服装商城管理系统
|
15天前
|
前端开发 JavaScript Java
基于Java+Springboot+Vue开发的大学竞赛报名管理系统
基于Java+Springboot+Vue开发的大学竞赛报名管理系统(前后端分离),这是一项为大学生课程设计作业而开发的项目。该系统旨在帮助大学生学习并掌握Java编程技能,同时锻炼他们的项目设计与开发能力。通过学习基于Java的大学竞赛报名管理系统项目,大学生可以在实践中学习和提升自己的能力,为以后的职业发展打下坚实基础。
38 3
基于Java+Springboot+Vue开发的大学竞赛报名管理系统
|
16天前
|
前端开发 JavaScript Java
基于Java+Springboot+Vue开发的蛋糕商城管理系统
基于Java+Springboot+Vue开发的蛋糕商城管理系统(前后端分离),这是一项为大学生课程设计作业而开发的项目。该系统旨在帮助大学生学习并掌握Java编程技能,同时锻炼他们的项目设计与开发能力。通过学习基于Java的蛋糕商城管理系统项目,大学生可以在实践中学习和提升自己的能力,为以后的职业发展打下坚实基础。
45 3
基于Java+Springboot+Vue开发的蛋糕商城管理系统
|
16天前
|
前端开发 JavaScript Java
基于Java+Springboot+Vue开发的美容预约管理系统
基于Java+Springboot+Vue开发的美容预约管理系统(前后端分离),这是一项为大学生课程设计作业而开发的项目。该系统旨在帮助大学生学习并掌握Java编程技能,同时锻炼他们的项目设计与开发能力。通过学习基于Java的美容预约管理系统项目,大学生可以在实践中学习和提升自己的能力,为以后的职业发展打下坚实基础。
29 3
基于Java+Springboot+Vue开发的美容预约管理系统
|
18天前
|
前端开发 JavaScript Java
基于Java+Springboot+Vue开发的房产销售管理系统
基于Java+Springboot+Vue开发的房产销售管理系统(前后端分离),这是一项为大学生课程设计作业而开发的项目。该系统旨在帮助大学生学习并掌握Java编程技能,同时锻炼他们的项目设计与开发能力。通过学习基于Java的房产销售管理系统项目,大学生可以在实践中学习和提升自己的能力,为以后的职业发展打下坚实基础。
36 3
基于Java+Springboot+Vue开发的房产销售管理系统
|
16天前
|
存储 网络协议 Java
Java NIO 开发
本文介绍了Java NIO(New IO)及其主要组件,包括Channel、Buffer和Selector,并对比了NIO与传统IO的优势。文章详细讲解了FileChannel、SocketChannel、ServerSocketChannel、DatagramChannel及Pipe.SinkChannel和Pipe.SourceChannel等Channel实现类,并提供了示例代码。通过这些示例,读者可以了解如何使用不同类型的通道进行数据读写操作。
Java NIO 开发
|
5天前
|
安全 Java API
Java 泛型在安卓开发中的应用
在Android开发中,Java泛型广泛应用于集合类、自定义泛型类/方法、数据绑定、适配器及网络请求等场景,有助于实现类型安全、代码复用和提高可读性。例如,结合`ArrayList`使用泛型可避免类型转换错误;自定义泛型类如`ApiResponse&lt;T&gt;`可处理不同类型API响应;RecyclerView适配器利用泛型支持多种视图数据;Retrofit结合泛型定义响应模型,明确数据类型。然而,需注意类型擦除导致的信息丢失问题。合理使用泛型能显著提升代码质量和应用健壮性。
|
3天前
|
安全 算法 Java
数据库信息/密码加盐加密 —— Java代码手写+集成两种方式,手把手教学!保证能用!
本文提供了在数据库中对密码等敏感信息进行加盐加密的详细教程,包括手写MD5加密算法和使用Spring Security的BCryptPasswordEncoder进行加密,并强调了使用BCryptPasswordEncoder时需要注意的Spring Security配置问题。
27 0
数据库信息/密码加盐加密 —— Java代码手写+集成两种方式,手把手教学!保证能用!
|
4天前
|
JSON Java 开发工具
Java服务端集成Google FCM推送的注意事项和实际经验
公司的app要上海外,涉及到推送功能,经过综合考虑,选择Google FCM进行消息推送。 查看一些集成博客和官方文档,看的似懂非懂,迷迷惑惑。本篇文章除了将我实际集成的经验分享出来,也会对看到的博客及其中产生的疑惑、注意事项一一评论。 从官方文档和众多博客中,你会发现Java集成FCM推送有多种实现方式,会让生产生文档很乱,不知作何选择的困惑。
22 0
|
3天前
|
存储 分布式计算 Java
Stream很好,Map很酷,但答应我别用toMap():Java开发中的高效集合操作
在Java的世界里,Stream API和Map集合无疑是两大强大的工具,它们极大地简化了数据处理和集合操作的复杂度。然而,在享受这些便利的同时,我们也应当警惕一些潜在的陷阱,尤其是当Stream与Map结合使用时。本文将深入探讨Stream与Map的优雅用法,并特别指出在使用toMap()方法时需要注意的问题,旨在帮助大家在工作中更高效、更安全地使用这些技术。
16 0