企业官网建设流程全解析

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/modules

2、解压 tar -zxvf datax.tar.gz -C /opt/installs

3、配置环境变量 vim /etc/profile.d/myenv.sh

export DATAX_HOME=/opt/installs/datax export PATH=$PATH:$DATAX_HOME/bin

source /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.json

table+column 模式

特点

1. 声明式读取，指定表和字段，不用手写 select DataX 自动生成 SQL：select empno,ename,job from emp
2. 配置了 splitPk="empno"，支持多 channel 并发分片读取speed.channel=3，DataX 会根据 empno 自动切分成 3 段并行拉取数据，速度更快
3. errorLimit 配置：脏数据条数 0，脏数据比例上限 0.02（2%），超出任务失败
4. streamwriter：print: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.json

querySql 自定义 SQL 模式

1. 手写完整 SQL，自由度高 可以写 join、where 过滤、函数、多表关联、子查询等复杂逻辑
2. speed.channel=1只能单通道串行读取，无法并发分片，大数据量表读取慢
3. 没有 errorLimit 脏数据控制
4. streamwriter：print: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.json

sliceRecordCount: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 先查看这个主键是否有数据，如果有直接删除并插入，如果没有，直接插入。