DATAX快速上手非常详细

article/2025/10/13 5:10:37

前言
         博主在工作的过程中有一天公司决定将数据迁移的新的项目上去，当我发现数据库中的表大于有4000多张表的时我顿时懵了下，这数据迁移人力物力消耗的也太大了吧(看DataX的设计)。所以我们可以借助阿里云开源的DataX来解决这个问题。
看完这篇掌握以下内容：

  • 什么是DataX
  • DataX的设计
  • Datax框架设计、运行原理
  • 快速入门

在这里插入图片描述

一、概论

1.1 什么是DataX

         DataX 是阿里巴巴开源的一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL、Oracle 等)、HDFS、Hive、ODPS、HBase、FTP 等各种异构数据源之间稳定高效的数据同步功能。

在这里插入图片描述

1.2 DataX 的设计

         为了解决异构数据源同步问题，DataX 将复杂的网状的同步链路变成了星型数据链路，DataX 作为中间传输载体负责连接各种数据源。当需要接入一个新的数据源的时候，只需要将此数据源对接到 DataX，便能跟已有的数据源做到无缝数据同步。
在这里插入图片描述

1.3 框架设计

在这里插入图片描述

  • Reader：数据采集模块，负责采集数据源的数据，将数据发给Framework。
  • Wiriter: 数据写入模块，负责不断向Framwork取数据，并将数据写入到目的端。
  • Framework:用于连接read和writer,作为两者的数据传输通道，并处理缓冲，流控，并发，数据转换等你核心技术问题。
    ##1.4 运行原理
    在这里插入图片描述
  • Job：单个作业的管理节点，负责数据清理、子任务划分、TaskGroup监控管理。
  • Task：由Job切分而来，是DataX作业的最小单元，每个Task负责一部分数据的同步工作。
  • Schedule：将Task组成TaskGroup，单个TaskGroup的并发数量为5。
  • TaskGroup：负责启动Task。

1.4 Datax所支持的渠道

类型数据源读者作家（写）文件
RDBMS关系型数据库的MySQL√√读，写
           甲骨文        √        √    读，写
SQL服务器√√读，写
PostgreSQL的√√读，写
DRDS√√读，写
通用RDBMS（支持所有关系型数据库）√√读，写
阿里云数仓数据存储ODPS√√读，写
美国存托凭证√写
开源软件√√读，写
OCS√√读，写
NoSQL数据存储OTS√√读，写
Hbase0.94√√读，写
Hbase1.1√√读，写
凤凰4.x√√读，写
凤凰5.x√√读，写
MongoDB√√读，写
蜂巢√√读，写
卡桑德拉√√读，写
无结构化数据存储文本文件√√读，写
的FTP√√读，写
HDFS√√读，写
弹性搜索√写
时间序列数据库OpenTSDB√读
技术开发局√√读，写

二、快速入门

2.1 环境搭建

下载地址： http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz
源码地址： https://github.com/alibaba/DataX

配置要求：

  • Linux
  • JDK(1.8以上 建议1.8) 下载
  • Python(推荐 Python2.6.X)下载
    安装：

1） 将下载好的datax.tar.gz上传到服务器的任意节点,我这里上传到node01上的/exprot/soft
2）解压到/export/servers/

[root@node01 soft]# tar -zxvf datax.tar.gz  -C ../servers/

3）运行自检脚本

[root@node01 soft]# cd /export/servers/datax
[root@node01 datax]# ll
总用量 36
drwxr-xr-x 2 62265 users 4096 10月  2 14:42 bin
drwxr-xr-x 2 62265 users 4096 10月  2 14:42 conf
drwxr-xr-x 2 62265 users 4096 10月  2 14:42 job
drwxr-xr-x 2 62265 users 4096 10月  2 14:42 lib
drwxr-xr-x 3 root  root  4096 8月   8 22:39 log
drwxr-xr-x 3 root  root  4096 8月   8 22:39 log_perf
drwxr-xr-x 4 62265 users 4096 10月 12 2019 plugin
drwxr-xr-x 2 62265 users 4096 10月  2 14:42 script
drwxr-xr-x 2 62265 users 4096 10月  2 14:42 tmp
[root@node01 datax]# bin/datax.py   job/job.json

出现以下结果说明你得环境没有问题
在这里插入图片描述

2.2从Stream流读取数流打印到控制台

查看官方提高的模板

[root@node01 datax]# python  bin/datax.py  -r streamreader -w streamwriter
DataX (DATAX-OPENSOURCE-3.0), From Alibaba !
Copyright (C) 2010-2017, Alibaba Group. All Rights Reserved.Please refer to the streamreader document:https://github.com/alibaba/DataX/blob/master/streamreader/doc/streamreader.mdPlease refer to the streamwriter document:https://github.com/alibaba/DataX/blob/master/streamwriter/doc/streamwriter.md,kPlease save the following configuration as a json file and  usepython {DATAX_HOME}/bin/datax.py {JSON_FILE_NAME}.json
to run the job.{"job": {"content": [{"reader": {"name": "streamreader","parameter": {"column": [],"sliceRecordCount": ""}},"writer": {"name": "streamwriter","parameter": {"encoding": "","print": true}}}],"setting": {"speed": {"channel": ""}}}
}

根据模板修改配置

[root@node01 datax]# vim job/stream2_stream.json
{"job": {"content": [{"reader": {"name": "streamreader","parameter": {"sliceRecordCount": 10,"column": [{"type": "long","value": "10"},{"type": "string","value": "hello，DataX"}]}},"writer": {"name": "streamwriter","parameter": {"encoding": "UTF-8","print": true}}}],"setting": {"speed": {"channel": 1}}}
}

执行脚本（部分日志）

[root@node01 datax]# bin/datax.py  job/stream2_stream.json	
2020-10-02 15:31:00.714 [job-0] INFO  JobContainer - PerfTrace not enable!
2020-10-02 15:31:00.714 [job-0] INFO  StandAloneJobContainerCommunicator - Total 10 records, 130 bytes | Speed 13B/s, 1                                                                                     records/s | Error 0 records, 0 bytes |  All Task WaitWriterTime 0.000s |  All Task WaitReaderTime 0.006s | Percentage 10                                                                                    0.00%
2020-10-02 15:31:00.715 [job-0] INFO  JobContainer -
任务启动时刻                    : 2020-10-02 15:30:50
任务结束时刻                    : 2020-10-02 15:31:00
任务总计耗时                    :                 10s
任务平均流量                    :               13B/s
记录写入速度                    :              1rec/s
读出记录总数                    :                  10
读写失败总数                    :                   0

2.3读取MySQL中的数据写入到HDFS

准备
创建数据库和表并加载测试数据

create database test;
use test;
create table c_s(id   varchar(100) null,c_id int          null,s_id varchar(20)  null
);
INSERT INTO test.c_s (id, c_id, s_id) VALUES ('123', 1, '201967');
INSERT INTO test.c_s (id, c_id, s_id) VALUES ('123', 2, '201967');
INSERT INTO test.c_s (id, c_id, s_id) VALUES ('123', 3, '201967');
INSERT INTO test.c_s (id, c_id, s_id) VALUES ('123', 5, '201967');
INSERT INTO test.c_s (id, c_id, s_id) VALUES ('123', 6, '201967');

查看官方提高的模板

[root@node01 datax]# bin/datax.py -r mysqlreader -w hdfswriterDataX (DATAX-OPENSOURCE-3.0), From Alibaba !
Copyright (C) 2010-2017, Alibaba Group. All Rights Reserved.Please refer to the mysqlreader document:https://github.com/alibaba/DataX/blob/master/mysqlreader/doc/mysqlreader.mdPlease refer to the hdfswriter document:https://github.com/alibaba/DataX/blob/master/hdfswriter/doc/hdfswriter.mdPlease save the following configuration as a json file and  usepython {DATAX_HOME}/bin/datax.py {JSON_FILE_NAME}.json
to run the job.{"job": {"content": [{"reader": {"name": "mysqlreader","parameter": {"column": [],"connection": [{"jdbcUrl": [],"table": []}],"password": "","username": "","where": ""}},"writer": {"name": "hdfswriter","parameter": {"column": [],"compress": "","defaultFS": "","fieldDelimiter": "","fileName": "","fileType": "","path": "","writeMode": ""}}}],"setting": {"speed": {"channel": ""}}}
}

根据官网模板进行修改

[root@node01 datax]# vim job/mysqlToHDFS.json
{"job": {"content": [{"reader": {"name": "mysqlreader","parameter": {"column": ["id","c_id","s_id"],"connection": [{"jdbcUrl": ["jdbc:mysql://node02:3306/test"],"table": ["c_s"]}],"password": "123456","username": "root"}},"writer": {"name": "hdfswriter","parameter": {"column": [{"name": "id","type": "string"},{"name": "c_id","type": "int"},{"name": "s_id","type": "string"}],"defaultFS": "hdfs://node01:8020","fieldDelimiter": "\t","fileName": "c_s.txt","fileType": "text","path": "/","writeMode": "append"}}}],"setting": {"speed": {"channel": "1"}}}
}

MySQL的参数介绍
在这里插入图片描述
HDFS参数介绍
在这里插入图片描述
运行脚本

[root@node01 datax]# bin/datax.py  job/mysqlToHDFS.json
2020-10-02 16:12:16.358 [job-0] INFO  HookInvoker - No hook invoked, because base dir not exists or is a file: /export/servers/datax/hook
2020-10-02 16:12:16.359 [job-0] INFO  JobContainer -[total cpu info] =>averageCpu                     | maxDeltaCpu                    | minDeltaCpu-1.00%                         | -1.00%                         | -1.00%[total gc info] =>NAME                 | totalGCCount       | maxDeltaGCCount    | minDeltaGCCount    | totalGCTime        | maxDeltaGCTime     | minDeltaGCTimePS MarkSweep         | 1                  | 1                  | 1                  | 0.245s             | 0.245s             | 0.245sPS Scavenge          | 1                  | 1                  | 1                  | 0.155s             | 0.155s             | 0.155s2020-10-02 16:12:16.359 [job-0] INFO  JobContainer - PerfTrace not enable!
2020-10-02 16:12:16.359 [job-0] INFO  StandAloneJobContainerCommunicator - Total 5 records, 50 bytes | Speed 5B/s, 0 records/s | Error 0 records, 0 bytes |  All Task WaitWriterTime 0.000s |  All Task WaitReaderTime 0.000s | Percentage 100.00%
2020-10-02 16:12:16.360 [job-0] INFO  JobContainer -
任务启动时刻                    : 2020-10-02 16:12:04
任务结束时刻                    : 2020-10-02 16:12:16
任务总计耗时                    :                 12s
任务平均流量                    :                5B/s
记录写入速度                    :              0rec/s
读出记录总数                    :                   5
读写失败总数                    :                   0

2.4 读取HDFS中的数据写入到MySQL

准备工作

create database test;
use test;
create table c_s2(id   varchar(100) null,c_id int          null,s_id varchar(20)  null
);

查看官方提供的模板

[root@node01 datax]# bin/datax.py -r hdfsreader -w mysqlwriterDataX (DATAX-OPENSOURCE-3.0), From Alibaba !
Copyright (C) 2010-2017, Alibaba Group. All Rights Reserved.Please refer to the hdfsreader document:https://github.com/alibaba/DataX/blob/master/hdfsreader/doc/hdfsreader.mdPlease refer to the mysqlwriter document:https://github.com/alibaba/DataX/blob/master/mysqlwriter/doc/mysqlwriter.mdPlease save the following configuration as a json file and  usepython {DATAX_HOME}/bin/datax.py {JSON_FILE_NAME}.json
to run the job.{"job": {"content": [{"reader": {"name": "hdfsreader","parameter": {"column": [],"defaultFS": "","encoding": "UTF-8","fieldDelimiter": ",","fileType": "orc","path": ""}},"writer": {"name": "mysqlwriter","parameter": {"column": [],"connection": [{"jdbcUrl": "","table": []}],"password": "","preSql": [],"session": [],"username": "","writeMode": ""}}}],"setting": {"speed": {"channel": ""}}}
}

根据官方提供模板进行修改

[root@node01 datax]# vim job/hdfsTomysql.json
{"job": {"content": [{"reader": {"name": "hdfsreader","parameter": {"column": ["*"],"defaultFS": "hdfs://node01:8020","encoding": "UTF-8","fieldDelimiter": "\t","fileType": "text","path": "/c_s.txt"}},"writer": {"name": "mysqlwriter","parameter": {"column": ["id","c_id","s_id"],"connection": [{"jdbcUrl": "jdbc:mysql://node02:3306/test","table": ["c_s2"]}],"password": "123456","username": "root","writeMode": "replace"}}}],"setting": {"speed": {"channel": "1"}}}
}

脚本运行

[root@node01 datax]# bin/datax.py job/hdfsTomysql.json[total cpu info] =>averageCpu                     | maxDeltaCpu                    | minDeltaCpu-1.00%                         | -1.00%                         | -1.00%[total gc info] =>NAME                 | totalGCCount       | maxDeltaGCCount    | minDeltaGCCount    | totalGCTime        | maxDeltaGCTime     | minDeltaGCTimePS MarkSweep         | 1                  | 1                  | 1                  | 0.026s             | 0.026s             | 0.026sPS Scavenge          | 1                  | 1                  | 1                  | 0.015s             | 0.015s             | 0.015s2020-10-02 16:57:13.152 [job-0] INFO  JobContainer - PerfTrace not enable!
2020-10-02 16:57:13.152 [job-0] INFO  StandAloneJobContainerCommunicator - Total 5 records, 50 bytes | Speed 5B/s, 0 records/s | Error 0 records, 0 bytes |  All Task WaitWriterTime 0.000s |  All Task WaitReaderTime 0.033s | Percentage 100.00%
2020-10-02 16:57:13.153 [job-0] INFO  JobContainer -
任务启动时刻                    : 2020-10-02 16:57:02
任务结束时刻                    : 2020-10-02 16:57:13
任务总计耗时                    :                 11s
任务平均流量                    :                5B/s
记录写入速度                    :              0rec/s
读出记录总数                    :                   5
读写失败总数                    :                   0

三、总结

         好了到这里今天的分享基本就结束了，上述给大家操作了从MySQL到HDFS/从HDFS到MySQL对于喜欢的小伙伴们可以参考1.4根据实际情况配置，细心的小伙伴会发现难道我每一使用都要配置他们？对于这问题有大佬开发了一个Datax-web可以根据页面的配置进行数据的同步这样就大大减少了我们的学习成本。我们下期见~~~


http://chatgpt.dhexx.cn/article/LpTp1fLj.shtml

相关文章

DataX及DataX-Web

DataX及DataX-Web

大数据Hadoop之——数据同步工具DataX数据采集工具-DataX datax详细介绍及使用 一、概述 DataX 是阿里云DataWorks数据集成的开源版本，在阿里巴巴集团内被广泛使用的离线数据同步工具/平台。DataX 实现了包括 MySQL、Oracle、OceanBase、SqlServer、Postgre、HDFS、…
阅读更多...
datax(4): datax.py解读

datax(4): datax.py解读

datax 直接使用py文件进行任务提交，今天读一读它 一、文件位置 原始文件位置在 xx/DataX/core/src/main/bin/下，datax项目打包后会将文件拷贝到 xx/DataX\target\datax\datax\bin 下。 core模块的pom.xml 指定‘拷贝’datax.py文件的方式maven-assembly…
阅读更多...
DataX使用指南

DataX使用指南

简介 DataX 是阿里巴巴集团内被广泛使用的离线数据同步工具/平台，实现包括 MySQL、Oracle、HDFS、Hive、OceanBase、HBase、OTS、ODPS 等各种异构数据源之间高效的数据同步功能。DataX采用了框架 插件 的模式，目前已开源，代码托管在github。…
阅读更多...
DataX

DataX

DataX的环境搭建以及简单测试 什么是DataX DataX 是一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL、Oracle等)、 HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能。 （这是一个单机多任务的ETL工具&#xff0…
阅读更多...
DataX 简介及架构原理

DataX 简介及架构原理

DataX 简介及架构原理 概述 DataX是阿里巴巴使用 Java 和 Python 开发的一个异构数据源离线同步工具 异构数据源：不同存储结构的数据源 致力于实现包括关系型数据库 MySQL、Oracle、SqlServer、Postgre、HDFS、Hive、ADS、HBase、TableStore(OTS)、MaxCompute(ODPS…
阅读更多...
DataX的使用与介绍（1）

DataX的使用与介绍（1）

一、什么是DataX？ DataX是阿里云商用产品DataWorks数据集成的开源版本，它是一个异构数据源的离线数据同步工具/平台（ETL工具）。DataX实现了包括Mysql，Oracle、OceanBase、Sqlserver，Postgre、HDFS、Hive、…
阅读更多...
DataX介绍

DataX介绍

DataX 是阿里开源的一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能。 DataX设计理念 DataX本身作为数据同步框架，将不同数据源的同步抽象为从源头…
阅读更多...
详解DataX及使用

详解DataX及使用

DataX概述 简介 DataX 是阿里巴巴开源的一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能。 支持数据源 DataX架构原理 设计理念 为了解决异构数据源同步问…
阅读更多...
使用 DataX 实现数据同步（高效的同步工具）

使用 DataX 实现数据同步（高效的同步工具）

DataX 使用介绍 前言一、DataX 简介1.DataX3.0 框架设计2.DataX3.0 核心架构 二、使用 DataX 实现数据同步1.Linux 上安装 DataX 软件2.DataX 基本使用3.安装 MySQL 数据库4.通过 DataX 实 MySQL 数据同步5.使用 DataX 进行增量同步 前言 我们公司有个项目的数据量高达五千万&…
阅读更多...
Transpose函数的用法

Transpose函数的用法

Transpose函数的用法 在CNN机器学习中，经常要用到transpose函数对多维数组进行转置操作，下面是我对函数的理解过程。 1.二维数组的转换 二维数组中，原数组的第0轴的行，转换成新数组第1轴的列； 2.三维数组 三维数组较…
阅读更多...
转置算子(transpose)的一种实现

转置算子(transpose)的一种实现

transpose算子也叫做permute算子，根据白嫖有道英汉大词典的结果，他俩都是转置，改变排列顺序的意思。 算法逻辑是： 通过当前输出的一维偏移量(offset)计算输入矩阵对应的高维索引 然后根据参数pos重新排列输出索引，进…
阅读更多...
论文笔记——TransPose

论文笔记——TransPose

目录 摘要 一、前言 二、相关工作 2.1 人体姿态估计 2.2 可解释性 三、TransPose 3.1 网络结构 3.2 分辨率设置 3.3 attentions是定位关键点的依赖 四、实验 4.1 COCO实验数据对比 4.2 迁移到MPII数据对比 4.3 消融实验​编辑 4.4 量化分析 五、总结 摘要 虽然基…
阅读更多...
numpy中的transpose函数使用

numpy中的transpose函数使用

二维矩阵的transpose函数 ： transpose（）简单来说，就相当于数学中的转置，在矩阵中，转置就是把行与列相互调换位置； 例如：随机生成一个三行五列的二维矩阵： arr np.aran…
阅读更多...
transpose()函数的理解

transpose()函数的理解

图1 输入如图1所示语句，输出如下： 图2 由以上两图说明transpose()函数的作用： 假设shape(z,x,y),在RGB图像中可以理解为z代表通道数，x代表图像的第几行，y代表图像的第几列，x和y组合而成所代表的像素构成…
阅读更多...
详解Python的transpose函数

详解Python的transpose函数

数组转置和换轴 import numpy as np >>> arr np.arange(16).reshape((2,2,4)) array([[[ 0, 1, 2, 3],[ 4, 5, 6, 7]],[[ 8, 9, 10, 11],[12, 13, 14, 15]]])>>> arr.transpose((1, 0, 2)) array([[[ 0, 1, 2, 3],[ 8, 9, 10, 11]],[[ 4, 5, …
阅读更多...
np.transpose

np.transpose

最近看代码的时候，老是出现np.transpose()这个用法，但是对其中的原理还是不甚了解，今天就来总结一下，以及这个用法对图像的结果及效果。 参数 a:输入数组 axis: int类型的列表，这个参数是可选的。默认情况下&#xff…
阅读更多...
np.transpose()函数详解

np.transpose()函数详解

1. 碰见 numpy.transpose 用于高维数组时挺让人费解，通过分析和代码验证，发现 transpose 用法还是很简单的。说白了就是映射坐标轴 2. 举个例子： x np.arange(12).reshape((2,3,2))创建一个2 * 3 * 2的数组： 使用 numpy.trans…
阅读更多...
【Python学习】transpose函数

【Python学习】transpose函数

shape:(batch_size * x * y ) 有batch_size个二维矩阵（x * y）相当于(z * x * y) 1. 多维数组的索引 import numpy as np # 创建 x np.arange(12).reshape((2,2,3)) print(x)# 得到三维数组 [[[ 0 1 2][ 3 4 5]][[ 6 7 8][ 9 10 11]]] # 相当于 b…
阅读更多...
最简单例子解释python的transpose函数

最简单例子解释python的transpose函数

目录 一，我们要弄清楚transpose的轴是什么意思？二，(x,y,z)的物理含义:三，transpose变换的例子四，代码验证五，关于经过了transpose变换之后，这个三维数组的形状是如何变化确定的？ 二维…
阅读更多...
转置（transpose）的理解

转置（transpose）的理解

目录 1 .T,适用于一、二维数组 arr.T #求转置 transpose 的原理其实是根据维度（shape）索引决定的，举个栗子： 2. 高维数组 3. swapaxes 转置可以对数组进行重置，返回的是源数据的视图（不会进行任何复制…
阅读更多...
推荐文章

Copyright @ 2022~2023