DataX的使用

DataX的介绍Github主页地址https://github.com/alibaba/DataX可以理解为国内版的Sqoop。但是比Sqoop要快Sqoop底层是MR(Map任务基于磁盘的DataX基于内存的所以速度比较快。DataX 是阿里巴巴集团内被广泛使用的离线数据同步工具/平台实现包括 MySQL、SQL Server、Oracle、PostgreSQL、HDFS、Hive、HBase、OTS、ODPS 等各种异构数据源之间高效的数据同步功能。关系型数据库的数据都是结构化的排列非常整齐非关系型数据库都是非结构化的非常的错乱。异构DataX3.0 框架设计datax中的数据导入导出是并行执行的并且是基于内存的所以比较快!支持的数据源有哪些几乎所有安装并测试1、上传到目录 /opt/modules2、解压 tar -zxvf datax.tar.gz -C /opt/installs3、配置环境变量 vim /etc/profile.d/myenv.shexport DATAX_HOME/opt/installs/datax export PATH$PATH:$DATAX_HOME/binsource /etc/profile安装后测试一下自带的案例编辑这个案例job.json文件运行一下datax.py job.json假如你运行报错如下datax.py job.json-bash: /opt/installs/datax/bin/datax.py: 权限不够chmod 777 /opt/installs/datax/bin/datax.py实战MySQLReader 案例创建一个表用于测试create database datax_test; use datax_test; CREATE TABLE emp( empno INT PRIMARY KEY, ename VARCHAR(50), job VARCHAR(50), mgr INT, hiredate DATE, sal DECIMAL(7,2), comm DECIMAL(7,2), deptno INT ) ; INSERT INTO emp VALUES(7369,SMITH,CLERK,7902,1980-12-17,800,NULL,20); INSERT INTO emp VALUES(7499,ALLEN,SALESMAN,7698,1981-02-20,1600,300,30); INSERT INTO emp VALUES(7521,WARD,SALESMAN,7698,1981-02-22,1250,500,30); INSERT INTO emp VALUES(7566,JONES,MANAGER,7839,1981-04-02,2975,NULL,20); INSERT INTO emp VALUES(7654,MARTIN,SALESMAN,7698,1981-09-28,1250,1400,30); INSERT INTO emp VALUES(7698,BLAKE,MANAGER,7839,1981-05-01,2850,NULL,30); INSERT INTO emp VALUES(7782,CLARK,MANAGER,7839,1981-06-09,2450,NULL,10); INSERT INTO emp VALUES(7788,SCOTT,ANALYST,7566,1987-04-19,3000,NULL,20); INSERT INTO emp VALUES(7839,KING,PRESIDENT,NULL,1981-11-17,5000,NULL,10); INSERT INTO emp VALUES(7844,TURNER,SALESMAN,7698,1981-09-08,1500,0,30); INSERT INTO emp VALUES(7876,ADAMS,CLERK,7788,1987-05-23,1100,NULL,20); INSERT INTO emp VALUES(7900,JAMES,CLERK,7698,1981-12-03,950,NULL,30); INSERT INTO emp VALUES(7902,FORD,ANALYST,7566,1981-12-03,3000,NULL,20); INSERT INTO emp VALUES(7934,MILLER,CLERK,7782,1982-01-23,1300,NULL,10);相关配置可参考配置的案例参考地址在github上找到你需要配置的数据库打开doc目录在job 文件夹创建一个文件 mysql2stream.jsontablecolumn 模式特点声明式读取指定表和字段不用手写 select DataX 自动生成 SQLselect empno,ename,job from emp配置了 splitPkempno支持多 channel 并发分片读取speed.channel3DataX 会根据 empno 自动切分成 3 段并行拉取数据速度更快errorLimit 配置脏数据条数 0脏数据比例上限 0.022%超出任务失败streamwriterprint:true控制台打印读出的数据{ job: { setting: { speed: { channel: 3 }, errorLimit: { record: 0, percentage: 0.02 } }, content: [ { reader: { name: mysqlreader, parameter: { username: root, password: root, column: [ empno, ename, job ], splitPk: empno, connection: [ { table: [ emp ], jdbcUrl: [ jdbc:mysql://bigdata001:3306/datax_test ] } ] } }, writer: { name: streamwriter, parameter: { print:true, encoding: UTF-8 } } } ] } }然后运行datax.py mysql2stream.jsonquerySql 自定义 SQL 模式手写完整 SQL自由度高 可以写 join、where 过滤、函数、多表关联、子查询等复杂逻辑speed.channel1只能单通道串行读取无法并发分片大数据量表读取慢没有 errorLimit 脏数据控制streamwriterprint:true控制台打印读出的数据在job 文件夹创建一个文件 mysql2stream2.json{ job: { setting: { speed: { channel:1 } }, content: [ { reader: { name: mysqlreader, parameter: { username: root, password: 123456, connection: [ { querySql: [ select empno,ename,job from emp; ], jdbcUrl: [ jdbc:mysql://bigdata001:3306/datax_test ] } ] } }, writer: { name: streamwriter, parameter: { print: true, encoding: UTF-8 } } } ] } }然后运行datax.py mysql2stream2.json读取mysql的数据将数据展示在控制台上。此时的stream其实就是控制台特殊说明如果你编写的json文件中需要用到字段类型必须指定DataX内部类型不要使用Mysql类型和java类型MySQLWriter 展示新建一个 emp2 表CREATE TABLE emp2( empno INT PRIMARY KEY, ename VARCHAR(50), job VARCHAR(50), mgr INT, hiredate DATE, sal DECIMAL(7,2), comm DECIMAL(7,2), deptno INT );编写json配置文件参考文档位置如下stream2mysql.jsonsliceRecordCount:10含义生成 10 条一模一样的该行记录{ job: { setting: { speed: { channel: 1 } }, content: [ { reader: { name: streamreader, parameter: { column: [ {value: 9870, type: long }, {value: SHAWN, type: string}, {value: BOSS, type: string}, {value: 1999-01-01, type: string}, {value: 99999, type: long} ], sliceRecordCount: 10 } }, writer: { name: mysqlwriter, parameter: { writeMode: insert, username: root, password: 123456, column: [empno, ename, job, hiredate, sal], connection: [ { jdbcUrl: jdbc:mysql://caijing:3306/datax, table: [emp] } ] } } } ] } }然后运行datax.py stream2mysql.json由于emp表中empno是主键所以10条数据只插入了一条。我们将writeMode: insert, 改成 replace,再次执行insert into 插入不判断直接插入replace into 先查看这个主键是否有数据如果有直接删除并插入如果没有直接插入。