【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即可。

结束语

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

目录
相关文章
|
29天前
|
XML 测试技术 API
利用C#开发ONVIF客户端和集成RTSP播放功能
利用C#开发ONVIF客户端和集成RTSP播放功能
910 122
|
14天前
|
人工智能 Java API
Java与大模型集成实战:构建智能Java应用的新范式
随着大型语言模型(LLM)的API化,将其强大的自然语言处理能力集成到现有Java应用中已成为提升应用智能水平的关键路径。本文旨在为Java开发者提供一份实用的集成指南。我们将深入探讨如何使用Spring Boot 3框架,通过HTTP客户端与OpenAI GPT(或兼容API)进行高效、安全的交互。内容涵盖项目依赖配置、异步非阻塞的API调用、请求与响应的结构化处理、异常管理以及一些面向生产环境的最佳实践,并附带完整的代码示例,助您快速将AI能力融入Java生态。
156 12
|
21天前
|
消息中间件 人工智能 Java
抖音微信爆款小游戏大全:免费休闲/竞技/益智/PHP+Java全筏开源开发
本文基于2025年最新行业数据,深入解析抖音/微信爆款小游戏的开发逻辑,重点讲解PHP+Java双引擎架构实战,涵盖技术选型、架构设计、性能优化与开源生态,提供完整开源工具链,助力开发者从理论到落地打造高留存、高并发的小游戏产品。
|
21天前
|
存储 Java 关系型数据库
Java 项目实战基于面向对象思想的汽车租赁系统开发实例 汽车租赁系统 Java 面向对象项目实战
本文介绍基于Java面向对象编程的汽车租赁系统技术方案与应用实例,涵盖系统功能需求分析、类设计、数据库设计及具体代码实现,帮助开发者掌握Java在实际项目中的应用。
41 0
|
2月前
|
安全 Java 数据库
Java 项目实战病人挂号系统网站设计开发步骤及核心功能实现指南
本文介绍了基于Java的病人挂号系统网站的技术方案与应用实例,涵盖SSM与Spring Boot框架选型、数据库设计、功能模块划分及安全机制实现。系统支持患者在线注册、登录、挂号与预约,管理员可进行医院信息与排班管理。通过实际案例展示系统开发流程与核心代码实现,为Java Web医疗项目开发提供参考。
109 2
|
2月前
|
JavaScript 安全 前端开发
Java开发:最新技术驱动的病人挂号系统实操指南与全流程操作技巧汇总
本文介绍基于Spring Boot 3.x、Vue 3等最新技术构建现代化病人挂号系统,涵盖技术选型、核心功能实现与部署方案,助力开发者快速搭建高效、安全的医疗挂号平台。
147 3
|
2月前
|
安全 Oracle Java
JAVA高级开发必备·卓伊凡详细JDK、JRE、JVM与Java生态深度解析-形象比喻系统理解-优雅草卓伊凡
JAVA高级开发必备·卓伊凡详细JDK、JRE、JVM与Java生态深度解析-形象比喻系统理解-优雅草卓伊凡
201 0
JAVA高级开发必备·卓伊凡详细JDK、JRE、JVM与Java生态深度解析-形象比喻系统理解-优雅草卓伊凡
|
3月前
|
并行计算 Java API
Java List 集合结合 Java 17 新特性与现代开发实践的深度解析及实战指南 Java List 集合
本文深入解析Java 17中List集合的现代用法,结合函数式编程、Stream API、密封类、模式匹配等新特性,通过实操案例讲解数据处理、并行计算、响应式编程等场景下的高级应用,帮助开发者提升集合操作效率与代码质量。
146 1
|
3月前
|
安全 Java API
Java 17 及以上版本核心特性在现代开发实践中的深度应用与高效实践方法 Java 开发实践
本项目以“学生成绩管理系统”为例,深入实践Java 17+核心特性与现代开发技术。采用Spring Boot 3.1、WebFlux、R2DBC等构建响应式应用,结合Record类、模式匹配、Stream优化等新特性提升代码质量。涵盖容器化部署(Docker)、自动化测试、性能优化及安全加固,全面展示Java最新技术在实际项目中的应用,助力开发者掌握现代化Java开发方法。
133 1