datax工具介绍及简单使用

这篇具有很好参考价值的文章主要介绍了datax工具介绍及简单使用。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

介绍

Datax是一个异构数据源离线同步工具,致力于实现包括关系数据库、HDFS、Hive、ODPS、Hbase等各种异构数据源之间稳定高效的数据同步功能

  • 设计理念

为了解决异构数据源同步问题,DataX将复杂的网状的同步链路变成了星型数据链路,DataX作为中间传输载体负责连接各种数据源。当需要接入一个新的数据源的时候,只需要将此数据源对接到DataX,便能跟已有的数据源做到无缝数据同步;
DataX在阿里巴巴集团内被广泛使用,承担了所有大数据的离线同步业务,并已 持续稳定运行了7年之久。目前每天完成同步8w多道作业,每日传输数据量超过300TB;

  • 框架设计

DataX本身作为离线数据同步框架,采用Framework + plugin架构构建。将数据源读取和写入抽象成为Reader/Writer插件,纳入到整个同步框架中,软件核心功能写入Framework 主体框架中。
datax工具介绍及简单使用,数据治理,大数据,数据同步,datax

主体框架为插件预留接口,如果后期需要什么新功能,可以再去开发插件实现,而主体框架无需改动。

  • 3.0核心架构

DataX 3.0 开源版本支持单机多线程模式完成同步作业运行,本小节按一个DataX作业 生命周期的时序图,从整体架构设计非常简要说明DataX各个模块相互关系
DataX作业生命周期的时序图
datax工具介绍及简单使用,数据治理,大数据,数据同步,datax

模块关系
DataX完成单个数据同步的作业,我们称之为Job,DataX接收到一个Job之后,将启 动一个进程来完成整个作业同步过程
DataX Job模块是单个作业的中枢管理节点, 承担了数据清理、子任务切分(将单一作业计算转化为多个子Task)、TaskGroup管理 等功能
DataX Job启动后,会根据不同的源端切分策略,将Job切分成多个小的Task(子任务),以便于并发执行
Task便是DataX作业的最小单元,每一个Task都会负责一部分数据的同步工作;
切分多个Task之后,DataX Job会调用Scheduler模块,根据配置的并发数据量,将拆分成的Task重新组合,组装成TaskGroup(任务组)
每一个TaskGroup负责并发运行完毕分配好的所有Task,默认单个任务组的并发数量为5(可配置) ;
每一个Task都由TaskGroup负责启动,Task启动后,会固定启动Reader— >Channel — >Writer的线程来完成任务同步工作
datax作业运行起来之后,job监控并等待多个taskgroup模块任务完成,完成之后job成功退出;否则,异常退出,进程退出值非0。

  • 调度流程

举例来说,用户提交了一个DataX作业,并且配置了20个并发,目的是将一个 100张分表的mysql数据同步到odps里面;
调度决策思路:
1)DataXJob根据分库分表切分成了100个Task;
2)根据20个并发,DataX计算共需要分配4个TaskGroup;
3)4个TaskGroup平分切分好的100个Task,每一个TaskGroup负责以5个并发共计运 行25个Task;
4)计算
100个Task,20个并发就是20个Channel;4(20 / 5 = 4) 个Taskgroup,每个Taskgroup内的25(100 / 4 = 25)个Task,能够同时运行的任务是20个Channel等于20个并发。

环境需求与安装

1)环境需求
Linux
JDK(1.8以上,推荐1.8)
Python(2或3都可以)

2)安装工具
下载后解压至本地某个目录,进入bin目录,即可运行同步作业
$ cd {YOUR_DATAX_HOME}/bin
$ python datax.py {YOUR_JOB.json}
自检脚本
$ python {YOUR_DATAX_HOME}/bin/datax.py {YOUR_DATAX_HOME}/job/job.json
如果报错
运行以下指令,再输入自检脚本指令
rm -rf {YOUR_DATAX_HOME}/plugin//._

具体就不详细介绍了。

入门案例

  • Stream -> Stream
    目的:打印输出10行 hello,世界
    步骤:
    1)打开Datax工具目录
    cd {YOUR_DATAX_HOME}

2)获取模板文件

python bin/datax.py -r streamreader -w streamwriter

3)修改

vim  job/stream2Stream.json

参数

column表示列

type表示该列的类型

value表示该列的值

column里面可以写多个列

sliceRecordCount:表示要打印多少次

encodding设置字符编码格式

print表示是否打印到控制台

setting

speed表示控制并发数

channel设置并发的数量

如果设置的print为true,则会打印slicRecordCount*channel次
如果是从mysql导入hdfs等其他操作,则会是真正代表并发数,而不是打印多少次

内容

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

4)运行

python bin/datax.py  job/stream2Stream.json
  • MySQL -> MySQL
    目的:MySQL数据库的t_emp表数据导入到另一个MySQL数据库的表中

步骤:获得模板文件
python bin/datax.py -r mysqlreader -w mysqlwriter

修改如下
vim job/mysql2Mysql.json

内容

{"job": {
"content": [{
"reader": {
"name": "mysqlreader","parameter": {
"column": ["*"],"splitPk": "emp_id",
"connection": [{"jdbcUrl": [jdbc:mysql://localhost:3306/demo?
useUnicode=true&characterEncoding=utf8"],"table": ["t_emp"]}]
"username": "root","password": "123456",}},
"writer": {
"name": "mysqlwriter","parameter": {
"column": ["*"],"connection": [{
"jdbcUrl": "jdbc:mysql://localhost:3306/democlone? useUnicode=true&characterEncoding=utf8","table": ["t_emp"]}],
"preSql": ["truncate t_emp"],
"session": ["set session sql_mode='ANSI'"],
"username": "root","password": "123456",
"writeMode": "insert"}}}],
"setting": {"speed": {"channel": "2"}}}}
//mysql端提前将表建好
//参数
//session sql_mode 数据校验模式

运行

python bin/datax.py job/mysql2Mysql.json
  • MySQL -> HDFS

目的:MySQL数据库中的t_emp表导入到HDFS中的/datax/mysql2hdfs/
步骤:
获得模板文件
修改如下
内容

{"job": {"content": [{
"reader": {"name": "mysqlreader","parameter": {
"column": [
"emp_id",
"dept_id",
"emp_name",
"emp_gender",
"emp_in_time"],
"connection": [{"jdbcUrl":["jdbc:mysql://localhost:3306/demo"],
"table": ["t_emp"]}],
"password": "123456","username": "root",
"where": ""}},
"writer": {"name": "hdfswriter",
"parameter": {"column": [{
type:"int",name:"emp_id"},{
type:"int",name:"dept_id"},{
type:"string",name:"emp_name"},{
type:"string",name:"emp_gender"},{
type:"date",name:"emp_in_time"}],
"compress":"",
"defaultFS": "hdfs://hadoop102:9000",
"fieldDelimiter": "\t",
"fileName": "mysqlToHdfs",
"fileType": "orc",
"path": "/datax/mysql2hdfs",
"writeMode": "append"}}}],
"setting": {"speed": {"channel": "2"}}}}

运行

python bin/datax.py job/mysql2Hdfs.json
  • MySQL -> Hive
    目的:MySQL数据库中的t_emp表导入到Hive数据仓库的orc_emp表
    步骤:类似上面的步骤

  • Hive -> MySQL

增量导入

目的:解决全量同步数据量较大时,同步被中断的问题
使用 DataX 进行全量同步和增量同步的唯一区别
增量同步需要使用 where 进 行条件筛选
关键参数
使用where参数或者querySql参数进行增量导入
步骤
修改
vim job/mysql2HiveIncrement.json
内容

{"job": {"content": [{
"reader": {"name": "mysqlreader","parameter": {
"column": [
"emp_id",
"dept_id",
"emp_name",
"emp_gender",
"emp_in_time"],
"connection": [{"jdbcUrl":["jdbc:mysql://localhost:3306/demo"],
"querySql":["select * from t_emp_from_hive
where emp_id > ${id}"]}],
"password": "123456","username": "root"}},
"writer": {"name": "hivewriter","parameter": {
"column": [{
type:"int",name:"emp_id"},{
type:"string",name:"emp_name"},{
type:"string",name:"emp_gender"},{
type:"date",name:"emp_in_time"}],
"defaultFS": "hdfs://hadoop102:9000",
"fieldDelimiter": "\t",
"fileName": "mysqlToHiveTEmp",
"fileType": "orc",
"path": "/user/hive/warehouse/demo.db/orc_emp", "writeMode": "append"}}}],
"setting": {"speed": {"channel": "3"}}}}

运行
赋值格式
-p “-DpropertyName=propertyValue -D…”
命令

$ python bin/datax.py job/mysql2HiveIncrement.json -p "-Did=20 -D"
$ python bin/datax.py job/mysql2HiveIncrement.json -p "-DstartId=0 -DendId=3"

Java执行Datax脚本

项目配置
环境
Windows 10+
安装 JDK1.8+、开发工具
安装 Python2.7.5+
下载解压datax.tar.gz
Maven本地仓库配置
解压Datax工具压缩包,提取两个jar文件 -core -common
安装两个依赖到Maven本地库

创建Maven项目
pom.xml文件中添加依赖

	<dependency>
            <groupId>com.alibaba.datax</groupId>
            <artifactId>datax-core</artifactId>
            <version>0.0.1</version>
        </dependency>
        <dependency>
        <groupId>com.alibaba.datax</groupId>
            <artifactId>datax-common</artifactId>
            <version>0.0.1</version>
        </dependency>
        <dependency>
            <groupId>commons-cli</groupId>
            <artifactId>commons-cli</artifactId>
            <version>1.4</version>
        </dependency>
        <dependency>
            <groupId>org.apache.httpcomponents</groupId>
            <artifactId>httpclient</artifactId>
            <version>4.5.13</version>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-io</artifactId>
            <version>1.3.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
            <version>3.12.0</version>
        </dependency>
        <dependency>
            <groupId>commons-lang</groupId>
            <artifactId>commons-lang</artifactId>
            <version>2.6</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.60</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.25</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.25</version>
        </dependency>
public class JobMysql2Mysql {

static Logger log = LoggerFactory.getLogger(Stream2Stream.class);

public static void main(String[] args) {
int exitCode = 0;
String dataXPath = "D:/MyProgramFiles/BigData/datax/";  String jobFilePath = dataXPath + "job/mysql2Mysql.json"; System.setProperty("datax.home", dataXPath);
String[] dataXArgs = {
"-job",  jobFilePath,
"-mode", "standalone",
"-jobid", "-1"
};
try {
Engine.entry(dataXArgs);
} catch(Throwable throwable) {
exitCode = -1;
log.error("\n\n经DataX智能分析,该任务最可能的错误原因::\n"
+ ExceptionTracker.trace(throwable));
}
System.exit(exitCode);
}
}
MySQL 增量导入
//目的
//迭代的方式将MySQL数据表中数据导入到另一个MySQL数据库表中
//使用Java程序迭代并传递参数的形式,完成MySQL数据库之间的数据同步

public class JobMysql2MysqlIncrement {
static Logger log = LoggerFactory.getLogger(Stream2Stream.class);

public static void main(String[] args) {
int exitCode = 0;
String dataXPath = "D:/MyProgramFiles/BigData/datax/";
String jobFilePath = dataXPath + "job/mysql2MysqlIncrement.json"; System.setProperty("datax.home", dataXPath);
String[] dataXArgs = {
"-job",  jobFilePath,
"-mode", "standalone",
"-jobid", "-1"
};
try {
// 数据表中总行数,可以使用JDBC获取
int rowTotal = 5;
// 迭代增量的形式,分批次导入数据
for(int i = 0; i < rowTotal; i = i + 3) {
System.setProperty("offset", String.valueOf(i));
System.setProperty("rows", "3");
Engine.entry(dataXArgs);
System.out.println("------------Job End------------");
}
} catch(Throwable throwable) {
exitCode = -1;
log.error("\n\n经DataX智能分析,该任务最可能的错误原因::\n" + ExceptionTracker.trace(throwable));
}
System.exit(exitCode);
}
}

注意
配置文件中需要写查询语句limit限制导入的数据量,java脚本中的参数才能传入,最后才能正常分批次运行

配置文件文章来源地址https://www.toymoban.com/news/detail-708609.html

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "column": ["*"],
                        "connection": [
                            {
                                "jdbcUrl": ["jdbc:mysql://192.168.2.102:3306/goods_sources?useUnicode=true&characterEncoding=utf8"],
                                "querySql":["select * from s_order limit ${offset},${rows}"]
                            }
                        ],
                        "username": "root",
                        "password": "123456"
                    }
                },
                "writer": {
                    "name": "mysqlwriter",
                    "parameter": {
                        "column": ["*"],
                        "connection": [
                            {
                                "jdbcUrl": "jdbc:mysql://192.168.2.102:3306/goods_targets?useUnicode=true&characterEncoding=utf8",    
                         "table": ["t_order"]
                            }
                        ],
                        "password": "123456",
                        "session": ["set session sql_mode='ANSI'"],
                        "username": "root",
                        "writeMode": "insert"
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": "2"
            }
        }
    }
}

补充

	DataX和sqoop的比较
		
	对于脏数据的处理
		在大量数据的传输过程中,必定会由于各种原因导致很多数据传输报错(比如类型转换错误),这种数据DataX认为就是脏数据。DataX目前可以实现脏数据精确过滤、识别、采集、展示,提供多种的脏数据处理模式
		Job支持用户对于脏数据的自定义监控和告警,包括对脏数据最大记录数阈值(record值)或者脏数据占比阈值(percentage值),当Job传输过程出现的脏数据大于用户指定的数量/百分比,DataX Job报错退出
		图中的配置的意思是当脏数据大于10条,或者脏数据比例达到0.05%,任务就会报错
	健壮的容错机制
		DataX作业是极易受外部因素的干扰,网络闪断、数据源不稳定等因素很容易让同步到一半的作业报错停止。因此稳定性是DataX的基本要求,在DataX 3.0的设计中,重点完善了框架和插件的稳定性
	丰富的数据转换功能
		DataX作为一个服务于大数据的ETL(Extract-Transform-Load 的缩写,即数据抽取、转换、装载的过程)工具,除了提供数据快照搬迁功能之外,还提供了丰富数据转换的功能,让数据在传输过程中可以轻松完成数据脱敏,补全,过滤等数据转换功能,另外还提供了自动groovy函数,让用户自定义转换函数

到了这里,关于datax工具介绍及简单使用的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 5、DataX(DataX简介、DataX架构原理、DataX部署、使用、同步MySQL数据到HDFS、同步HDFS数据到MySQL)

    1.1 DataX概述 源码地址:https://github.com/alibaba/DataX 1.2 DataX支持的数据源 DataX目前已经有了比较全面的插件体系,主流的RDBMS数据库、NOSQL、大数据计算系统都已经接入,目前支持数据如下图。 2.1 DataX设计理念 为了解决异构数据源同步问题,DataX将复杂的网状的同步链路变成了星

    2024年02月11日
    浏览(54)
  • 阿里云开源离线同步工具DataX3.0,用于数据仓库、数据集市、数据备份

    DataX是阿里云开源的一款离线数据同步工具,支持多种数据源和目的地的数据同步,包括但不限于MySQL、Oracle、HDFS、Hive、ODPS等。它可以通过配置文件来定义数据源和目的地的连接信息、数据同步方式、数据过滤等,从而实现数据的高效、稳定、可扩展的同步。 例如,如果您

    2024年02月10日
    浏览(53)
  • 数据同步工具调研选型:SeaTunnel 与 DataX 、Sqoop、Flume、Flink CDC 对比

    Apache SeaTunnel 是一个非常易用的超高性能分布式数据集成产品,支持海量数据的离线及实时同步。每天可稳定高效同步万亿级数据,已应用于数百家企业生产,也是首个由国人主导贡献到 Apache 基金会的数据集成顶级项目。 SeaTunnel 主要解决数据集成领域的常见问题: * 数据源

    2024年02月04日
    浏览(50)
  • 00-开源离线同步工具DataX3.0重磅详解!

    DataX 是一个异构数据源离线同步工具,致力于实现包括关系型数据库(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能。 为了解决异构数据源同步问题,DataX将复杂的网状的同步链路变成了星型数据链路,DataX作为中间传输载体负责连接

    2024年02月08日
    浏览(42)
  • DataX-阿里开源离线同步工具在Windows上实现Sqlserver到Mysql全量同步和增量同步

    Kettle-开源的ETL工具集-实现SqlServer到Mysql表的数据同步并部署在Windows服务器上: Kettle-开源的ETL工具集-实现SqlServer到Mysql表的数据同步并部署在Windows服务器上_etl实现sqlserver报表服务器_霸道流氓气质的博客-CSDN博客 上面讲过Kettle的使用,下面记录下阿里开源异构数据源同步工具

    2024年02月08日
    浏览(48)
  • DataX同步达梦数据到HDFS

    DataX同步达梦数据到HDFS 1、 前提条件 安装达梦数据库客户端 安装Python3.5 以上 导入dmPython模块 导入dmPython流程 在达梦数据库客户端 driverspythondmPython 这个路径下执行 ● 如果报错在PATH中加入 E:dmdbmsbin 达梦数据库的安装路径,并重新装载dmPython Traceback (most recent call last): ●

    2024年02月03日
    浏览(36)
  • DataX实现Mysql与ElasticSearch(ES)数据同步

    jdk1.8及以上 python2 查看是否安装成功 查看python版本号,判断是否安装成功 在datax/job下,json格式,具体内容及主要配置含义如下 mysqlreader为读取mysql数据部分,配置mysql相关信息 username,password为数据库账号密码 querySql:需要查询数据的sql,也可通过colums指定需要查找的字段(

    2024年02月05日
    浏览(59)
  • 阿里巴巴开源DataX全量同步多个MySQL数据库

    上次 写了阿里巴巴高效的离线数据同步工具DataX: https://mp.weixin.qq.com/s/_ZXqA3H__Kwk-9O-9dKyOQ 安装DataX这个开源工具,并且同步备份了几张数据表。但是发现一个问题,就是每张表都需要单独写一个 job。如果数据表有几百张是不是要写几百个,这个不太现实了。 正当一筹莫展之际

    2024年02月02日
    浏览(64)
  • Centos7.9通过datax-web2.0_用Datax3.0进行增量同步_增量删除_数据更新---大数据之DataX工作笔记006

     1.注意这里的增量同步,不像之前用的DBsyncer或者是,NIFI中的利用binlog的形式,实现真正的实时的数据同步.  2.这里的增量是,指定通过ID,或者时间来进行增量,比如大于2023-07-03 11:44:56的数据仅仅同步这个,或者是,id大于多少的这样,这里建议用时间,因为如果有id用的字符串咋弄来

    2024年02月10日
    浏览(39)
  • DolphinScheduler 调度 DataX 实现 MySQL To ElasticSearch 增量数据同步实践

    基于SQL查询的 CDC(Change Data Capture): 离线调度查询作业,批处理。把一张表同步到其他系统,每次通过查询去获取表中最新的数据。也就是我们说的基于SQL查询抽取; 无法保障数据一致性,查的过程中有可能数据已经发生了多次变更; 不保障实时性,基于离线调度存在天然的

    2024年02月03日
    浏览(44)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包