大数据技术之DataX DataX插件开发
Posted 脚丫先生
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了大数据技术之DataX DataX插件开发相关的知识,希望对你有一定的参考价值。
文章目录
一、背景
DataX 是阿里巴巴开源的一个异构数据源离线同步工具,致力于实现包括关系型数据库(mysql、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能。但是随着业务需求的增加,Datax自带的插件逐渐不满足,因此需要进行二次开发新的插件。基于此,本文对datax插件的二次开发步骤进行详细的说明,希望能帮助到大家。
二、基于java的本地测试datax
2.1 github上下载datax的源代码
网址: https://github.com/alibaba/DataX
2.2 datax代码导入idea
1.在此,把从Git上下载的datax代码,本地解压导入idea,如下图所示:
2. 基于java的本地测试
利用本地的maven进行编译datax源代码,打包命令如下:
mvn -U clean package assembly:assembly -Dmaven.test.skip=true
等待打包完成…即生成target目录:
3. 利用核心类,Core模块下的Engine进行本地测试:
4 修改Engine.java代码,因为这里是datax的入口。
首先设置系统变量,也就是datax的目录路径:
System.setProperty("datax.home","F:\\\\gx\\\\DataX\\\\target\\\\datax\\\\datax");
其次设置job路径和一些参数。
String[] datxArgs = {"-job", "F:\\\\gx\\\\DataX\\\\target\\\\datax\\\\datax\\\\job\\\\job.json", "-mode", "standalone", "-jobid", "-1"};
(datax目录路径和job路径用的是打包生成的target目录里的datax) 进行本地测试的时候,修改job.json即可。
测试: 直接点击main方法运行。
三、docker安装南大通用数据库GBase和GBase 8a
3.1 docker安装Gbase 8a
1 第一步 环境准备
确认机器上已经安装了Docker环境
2 第二步 在安装好docker的服务器上进行下载GBase 8a镜像
$ docker pull shihd/gbase8a:1.0
3 第三步 创建并启动容器
$ docker run --name 容器名 -itd -p5258:5258 shihd/gbase8a:1.0
GBase8a数据库信息
DB: gbase
User: root
Password: root
Port: 5258
3.2 docker安装Gbase 8s
1.查找GBase 8s 镜像版本
[root@localhost ~]# docker search gbase8s
2. 拉取GBase 8s镜像
[root@localhost ~]# docker pull liaosnet/gbase8s:3.3.0_2_amd64
3 运行容器
docker run --name 容器名 -p 19088:9088 -itd liaosnet/gbase8s:3.3.0_2_amd64
4 安装成功
通过docker ps 命令查看安装是否成功。并且进入容器:docker exec -i -t 容器名 /bin/bash
进入容器,查看数据库状态:
在则查看gbase 8s的实例服务:
四、南大通用数据库GBase 8s To GBase 8a
4.1 GBase 8s的reader读插件开发(writer同理)
1 新建gbase8sreader模块
2 复制mysqlreader的assembly包和resource包下的plugin.json、plugin_job_template.json
把复制的放入gbase8sreader模块相应的位置。
3. 修改复制过来的文件。
package.xml
<assembly
xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
<id></id>
<formats>
<format>dir</format>
</formats>
<includeBaseDirectory>false</includeBaseDirectory>
<fileSets>
<fileSet>
<directory>src/main/resources</directory>
<includes>
<include>plugin.json</include>
<include>plugin_job_template.json</include>
</includes>
<outputDirectory>plugin/reader/gbase8sreader</outputDirectory>
</fileSet>
<fileSet>
<directory>target/</directory>
<includes>
<include>gbase8sreader-0.0.1-SNAPSHOT.jar</include>
</includes>
<outputDirectory>plugin/reader/gbase8sreader</outputDirectory>
</fileSet>
</fileSets>
<dependencySets>
<dependencySet>
<useProjectArtifact>false</useProjectArtifact>
<outputDirectory>plugin/reader/gbase8sreader/libs</outputDirectory>
<scope>runtime</scope>
</dependencySet>
</dependencySets>
</assembly>
plugin.json
{
"name": "gbase8sreader",
"class": "com.cetc.datax.plugin.reader.gbase8sreader.Gbase8sReader",
"description": "useScene: test. mechanism: use datax framework to transport data from txt file. warn: The more you know about the data, the less problems you encounter.",
"developer": "cetc"
}
注意:“class” 为我们后面需要编写的代码类名
plugin_job_template.json
{
"name": "gbase8sreader",
"parameter": {
"username": "",
"password": "",
"column": [],
"connection": [
{
"jdbcUrl": [],
"table": []
}
],
"where": ""
}
}
4 新建插件代码包
Gbase8sReader代码如下所示:
package com.cetc.datax.plugin.reader.gbase8sreader;
import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.spi.Reader;
import com.alibaba.datax.common.util.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
import java.util.List;
import com.alibaba.datax.plugin.rdbms.reader.Constant;
import com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader;
/**
* @author : shujuelin
* @date : 16:08 2021/10/11
*/
public class Gbase8sReader extends Reader {
private static final DataBaseType DATABASE_TYPE = DataBaseType.Gbase8s;
public static class Job extends Reader.Job {
private static final Logger LOG = LoggerFactory
.getLogger(Job.class);
private Configuration originalConfig = null;
private CommonRdbmsReader.Job commonRdbmsReaderJob;
@Override
public void init() {
this.originalConfig = super.getPluginJobConf();
Integer userConfigedFetchSize = this.originalConfig.getInt(Constant.FETCH_SIZE);
if (userConfigedFetchSize != null) {
LOG.warn("对 gbasereader 不需要配置 fetchSize, gbasereader 将会忽略这项配置. 如果您不想再看到此警告,请去除fetchSize 配置.");
}
this.originalConfig.set(Constant.FETCH_SIZE, Integer.MIN_VALUE);
this.commonRdbmsReaderJob = new CommonRdbmsReader.Job(DATABASE_TYPE);
this.commonRdbmsReaderJob.init(this.originalConfig);
}
@Override
public void preCheck(){
init();
this.commonRdbmsReaderJob.preCheck(this.originalConfig,DATABASE_TYPE);
}
@Override
public List<Configuration> split(int adviceNumber) {
return this.commonRdbmsReaderJob.split(this.originalConfig, adviceNumber);
}
@Override
public void post() {
this.commonRdbmsReaderJob.post(this.originalConfig);
}
@Override
public void destroy() {
this.commonRdbmsReaderJob.destroy(this.originalConfig);
}
}
public static class Task extends Reader.Task {
private Configuration readerSliceConfig;
private CommonRdbmsReader.Task commonRdbmsReaderTask;
@Override
public void init() {
this.readerSliceConfig = super.getPluginJobConf();
this.commonRdbmsReaderTask = new CommonRdbmsReader.Task(DATABASE_TYPE,super.getTaskGroupId(), super.getTaskId());
this.commonRdbmsReaderTask.init(this.readerSliceConfig);
}
@Override
public void startRead(RecordSender recordSender) {
int fetchSize = 1000; //this.readerSliceConfig.getInt(Constant.FETCH_SIZE);
this.commonRdbmsReaderTask.startRead(this.readerSliceConfig, recordSender,
super.getTaskPluginCollector(), fetchSize);
}
@Override
public void post() {
this.commonRdbmsReaderTask.post(this.readerSliceConfig);
}
@Override
public void destroy() {
this.commonRdbmsReaderTask.destroy(this.readerSliceConfig);
}
}
}
注意
代码里的:private static final DataBaseType DATABASE_TYPE = DataBaseType.Gbase8s;
需要在
进行修改,增加自己的数据源类型。
Gbase("gbase","com.gbase.jdbc.Driver"),
Gbase8s("gbase8s","com.gbasedbt.jdbc.Driver");
在DataBaseType类中针对reader的数据源,要修改appendJDBCSuffixForReader方法:
同理: 编写writer插件时候也需要在相应的方法里appendJDBCSuffixForWriter进行修改。
5 打包运行
将其他模块注释只留下公共模块和自己的项目模块。最外层的pom
在最外层的package.xml加上下面这个
<fileSet>
<directory>gbase8sreader/target/datax/</directory>
<includes>
<include>**/*.*</include>
</includes>
<outputDirectory>datax</outputDirectory>
</fileSet>
五、南大通用gbase可视化工具
本文使用的客户端连接工具为:
5.1 南大通用GBase 8a的连接配置
5.2 南大通用GBase 8s的连接配置
以上是关于大数据技术之DataX DataX插件开发的主要内容,如果未能解决你的问题,请参考以下文章