数据集成工具 ---- datax 3.0

1、datax:

        是一个异构数据源离线同步工具,致力于实现关系型数据库(mysql、oracle等)hdfs、hive、hbase等各种异构数据源之间的数据同步

2、参考网址文献:

https://github.com/alibaba/DataX/blob/master/introduction.mdhttps://github.com/alibaba/DataX/blob/master/userGuid.mdhttps://github.com/alibaba/DataX/blob/master/introduction.md

3、Datax的框架设计:

Datax作为离线数据同步工具,主要的是采用了Framework+plugin架构构成,将数据源的读数据和写数据封装成对应的Reader和Writer插件,纳入到整体的同步框架中。

        1、Reader:作为数据的采集模块,负责采集数据源的数据,将数据发送给Framework

        2、Writer:作为数据写入模块,负责不断的向Framework取出数据,将数据写入到对应的目的端

        3、Framework:用于连接Reader和Writer,作为两者的数据传输通道,并处理缓冲、数据转换等核心技术问题。

4、Datax的核心架构:

Datax3.0 开源版本是支持单机多线程来完成同步作业运行,因为底层是使用java做开发。整体的架构:

模块的核心介绍:

        1、Datax完成单个数据同步做作业,被称之为job,Datax接收到一个job时就会启动一个进程来完成数据同步工作,所以Datax job 模块是单个作业的中枢管理中心,主要是承担数据清理,子任务切分、TaskGroup 管理。

        2、当Datax启动后,Datax job会根据不同的源数据将job切分成不同的Task,所以Task是Datax的最小作业单位,每一个Task都会负责一部分的数据同步。

        3、切分成多个Task后,Datax job 就会调用scheduler模块,根据配置的并发数量,将拆分的Task重新组合,组装成TaskGroup,每一个TaskGroup都负责一定的Task任务的执行,默认TaskGroup并发数量数5个。

        4、每一个Task都是由TaskGroup所监控执行启动,每一个Task启动后都会按照Reader---Channel---Writer的执行顺序执行。

        5、当任务启动后,Datax job就会监控所有的TaskGroup的执行情况,当所有的TaskGroup任务完成后,job就会退出,当出现异常,就会异常退出并且进程退出值非0.

5、Datax的核心优势:

        1、可靠的数据质量监控

        2、丰富的数据转换功能

        3、精准的控制速度

        4、容错机制:

                1、线程内部重试

                        DataX的核心插件都经过团队的全盘review,不同的网络交互方式都有不同的重试策略。

                2、线程级别重试

                        目前DataX已经可以实现TaskFailover,针对于中间失败的Task,DataX框架可以做到整个Task级别的重新调度。

        5、极简的体验

6、Datax与Sqoop的区别:
功能Dataxsqoop
运行模式单进程多线程MR
分布式是不支持分布式支持
流控需要定制
统计信息支持不支持,分布式的数据收集不方便
数据校验只有core部分有校验功能不支持,分布式的数据收集不方便
监控需要定制需要定制
7、Datax部署:

1、下载jar包:

        下载路径:https://github.com/alibaba/DataX

2、解压文件,配置环境变量:  

#解压jar包
tar -zxvf datax.tar.gz#配置环境变量:
vim /etc/profileexport   DATAX_HOME=/user/loacl/soft/datax
export   PATH=.:$PATH:$DATAX_HOME/bin#配置好环境变量,让配置文件生效
source /etc/profile

3、使执行文件拥有执行权:

添加执行权:chmod +x  data.py
8、Datax的使用:
在datax中会自动的生成模板的命令:datax.py -r streamreader -w streamwriter
        1、streamreader  to  streamwriter,数据打印在控制台上面

参数说明:

"sliceRecordCount": 100  #指定打印的个数"channel": 1  #指定并发度
#创建json文件:vim streamreadertostreamwriter.json{"job": {"content": [{"reader": {"name": "streamreader", "parameter": {"column": [{"type":"string","value":"wyz"},{"type":"int","value":"18"}], "sliceRecordCount": 100  #指定打印的个数}}, "writer": {"name": "streamwriter", "parameter": {"encoding": "", "print": true}}}], "setting": {"speed": {"channel": 1  #指定并发度}}}
}#脚本执行命令:
datax.py streamreadertostreamwriter.json
        2、mysql to mysql
1、可以通过命令获取模板:
datax.py  -r mysqlreader -w mysqlwriter 2、可以通过github上的模板进行编写:分别是mysqlreader和mysqlwriter,参数会比较详细3、在插入数据的需要注意是在将数据写入的时候如果出现在数据,那么此时可能是创建的表出了问题例如:表中的某个字段是主键,主键唯一
vim mysqlreaderTomysqlwriter.json{"job": {"content": [{"reader": {"name": "mysqlreader", "parameter": {"column": ["id","name","age","clazz","gender"], "connection": [{"jdbcUrl": ["jdbc:mysql://192.168.226.1:3306/bigdata25"], "table": ["stu"]}], "password": "123456", "username": "root", "where": ""      #不是必须要写的,作用是可以在读数据时进行一次过滤}}, "writer": {"name": "mysqlwriter", "parameter": {"column": ["id","name","age","clazz","gender"], "connection": [{"jdbcUrl": "jdbc:mysql://192.168.226.1:3306/bigdata25", "table": ["data_test"]}], "password": "123456", "preSql": [],   #不是必须写的,作用是再写入数据前可以执行该sql"session": [], "username": "root", "writeMode": "insert"   #必选,指定数据写入的模式,分成三种:insert(一般默认)、replace、update}}}], "setting": {"speed": {"channel": 1}}}
}#执行脚本:
datax.py   mysqlreaderTomysqlwriter.json

 将数据写入到mysql时,写入的表是需要提前创建的。

        3、mysql to hdfs

参数解释:

使用datax的时候hdfswriter只支持两种文件形式,分别是text和orc"fileType": "text",  #支持两种方式:text和orc,text表示的是textfile,orc表示的orcfile"compress": "", #指定文件的压缩形式,不指定代表不用压缩,text支持的压缩方式:gzip,bzip2,orc支持的压缩方式有NONE和SNAPPY"writeMode": "append"   #表示的是数据在写入的操作,分成三种:append,写入前不做任何处理  nonconflit 如果文件存在,直接报错  truncate:如果文件存在,那就先删除在写入"path": "/bigdata25/datax/datax_mysqltohdfs/"  #文件不存在是需要提前创建的
vim mysqlTohdfs.json{"job": {"content": [{"reader": {"name": "mysqlreader", "parameter": {"column": ["id","name","age","clazz","gender"], "connection": [{"jdbcUrl": ["jdbc:mysql://master:3306/bigdata25"], "table": ["stu"]}], "password": "123456", "username": "root", "where": ""}}, "writer": {"name": "hdfswriter", "parameter": {"column": [{"name":"id","type":"string"},{"name":"stu_name","type":"string" },{"name":"age","type":"string"},{"name":"clazz","type":"string"},{"name":"gender","type":"string" }], "compress": "", "defaultFS": "hdfs://master:9000", "fieldDelimiter": ",", "fileName": "stu_mysqltohdfs", "fileType": "text",  "path": "/bigdata25/datax/datax_mysqltohdfs/", "writeMode": "append"                         												}}}], "setting": {"speed": {"channel": 1}}}
}#执行脚本:
datax.py mysqlTohdfs.json
        4、mysql to hive

 原理思想:

实际上还是将数据存入到hdfs上面,hive通过记录元数据信息来获取数据
原理:创建好hive表在保存在hdfs上,是有文件路径,然后通过写入到指定的hdfs文件路径就能将数据写入到hive表中
当开启hive的时候,在hive中创建的表会默认的存储hdfs的/user/hive/warehouse/目录下

前期准备:

前期准备:
启动hive(后台启动):nohup hive --service metastore &
连接hive:hive创建hive表(在没有说明的情况下一般在都是创建一个外部表)
创建一个datax数据库:create database datax;
切换数据库:use datax
创建外部表:create external table if not exists datax_mysqltohive(id string,name string,age int,clazz string,gender string)
row format delimited  fields terminated  by ',' stored as textfile

编写脚本:

vim mysqlTohive.json{"job": {"content": [{"reader": {"name": "mysqlreader", "parameter": {"column": ["id","name","age","clazz","gender"], "connection": [{"jdbcUrl": ["jdbc:mysql://master:3306/bigdata25"], "table": ["stu"]}], "password": "123456", "username": "root", "where": ""}}, "writer": {"name": "hdfswriter", "parameter": {"column": [{"name":"id","type":"string"},{"name":"stu_name","type":"string" },{"name":"age","type":"string"},{"name":"clazz","type":"string"},{"name":"gender","type":"string" }], "compress": "", "defaultFS": "hdfs://master:9000", "fieldDelimiter": ",", "fileName": "datax_mysqltohive1", "fileType": "text", "path": "/user/hive/warehouse/datax.db/datax_mysqltohive", "writeMode": "append"}}}], "setting": {"speed": {"channel": 1}}}
}#执行脚本命令:
datax.py  mysqlTohive.json
        5、mysql to hbase(Hbase11XWriter)

参数解释:

"mode": "normal" 写hbase的模式,目前只支持normal模式"hbaseConfig"  {"hbase.zookeeper.quorum": "***"} 描述:连接HBase集群需要的配置信息,JSON格式"table": "writer" 表的名称,大小写比较敏感"encoding" 编码方式"rowkeyColumn" 描述:要写入的hbase的rowkey列。index:指定该列对应reader端column的索引,从0开始,若为常量index为-1;type:指定写入数据类型,用于转换HBase byte[];value:配置常量,常作为多个字段的拼接符。hbasewriter会将rowkeyColumn中所有列按照配置顺序进行拼接作为写入hbase的rowkey,不能全为常量"versionColumn" 表示指定写入hbase的时间戳,支持当前时间、指定时间列、指定时间

前期准备:

前期准备:启动zookeeper:zkServer.sh start(每一个节点上都是需要启动的)
查看zk的状态:zkServer.sh status 启动hbase:start-hbase.sh连接hbase:sqlline.py master,node1,node2进入hbase的客户端:hbase shell hbase中查看表的命令:!table
退出命令 !quit
查看表:list
在hbase中创建创建表,指定表名和列簇:create 'student','cf1'

编写脚本:

vim mysqlTohbase.json{"job": {"content": [{"reader": {"name": "mysqlreader", "parameter": {"column": ["id","name","age","clazz","gender"], "connection": [{"jdbcUrl": ["jdbc:mysql://master:3306/bigdata25"], "table": ["stu"]}], "password": "123456", "username": "root", "where": ""}}, "writer": {"name": "hbase11xwriter","parameter": {"hbaseConfig": {"hbase.zookeeper.quorum": "master:2181,node1:2181,node2:2181"},"table": "NEW_STU","mode": "normal","rowkeyColumn": [{"index":0,"type":"string"},{"index":-1,"type":"string","value":"_"}],"column": [{"index":1,"name": "cf1:name","type": "string"},{"index":2,"name": "cf1:age","type": "int"},{"index":3,"name": "cf1:clazz","type": "string"},{"index":4,"name": "cf1:gender","type": "string"}],"versionColumn":{"index": -1,"value":"123456789"},"encoding": "utf-8"}}}],"setting": {"speed": {"channel": 1}}  }
}#执行脚本
datax.py mysqlTohbase.json
        6、mysql增量同步数据到hive中。

最主要的工作就是在原先的mysql数据导入的hive中的基础上进行where过滤

vim mysqlTohive.json{"job": {"content": [{"reader": {"name": "mysqlreader", "parameter": {"column": ["id","name","age","clazz","gender"], "connection": [{"jdbcUrl": ["jdbc:mysql://master:3306/bigdata25"], "table": ["stu"]}], "password": "123456", "username": "root", "where": "id >20"}}, "writer": {"name": "hdfswriter", "parameter": {"column": [{"name":"id","type":"string"},{"name":"stu_name","type":"string" },{"name":"age","type":"string"},{"name":"clazz","type":"string"},{"name":"gender","type":"string" }], "compress": "", "defaultFS": "hdfs://master:9000", "fieldDelimiter": ",", "fileName": "datax_mysqltohive1", "fileType": "text", "path": "/user/hive/warehouse/datax.db/datax_mysqltohive", "writeMode": "append"}}}], "setting": {"speed": {"channel": 1}}}
}#执行脚本命令:
datax.py  mysqlTohive.json
9、在使用datax过程中出现的错误:

1、配置文件出现错误,脚本不完整:

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.hqwc.cn/news/541343.html

如若内容造成侵权/违法违规/事实不符,请联系编程知识网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Vue必须知道的知识2

v-on的指令 后可以写函数和代码和函数调用 v-on:clickfn/fn()/console.log(1)这三种,当执行new Vue()会去,add...事件,如果后面是函数,直接给,如果是代码/调用的,直接把内的判断。如果不是函数&#xff0…

Java new String getBytes 使用utf8静态枚举 StandardCharsets.UTF_8

Java new String 使用utf8静态枚举 用 StandardCharsets.UTF_8 new String(byteArray "utf8")或 getByte("utf8")要捕获异常 new String(byteArray, StandardCharsets.UTF_8) 或 getBytes(StandardCharsets.UTF_8) 就不用捕获异常了 StandardCharsets.U…

国产化系统查看配置

1、查看电脑系统/CPU/GPU等 1.1 开始 -》工具箱 可以查看CPU/GPU/操作系统等详细信息。 1.2 查看系统CPU 在计算机右键属性,可以查看到系统(统信UOS、银河麒麟);CPU(x86兆芯、x86海光、ARM鲲鹏、ARM phytium飞腾、L…

Linux编译器gcc/g++的功能与使用

一、程序的生成 首先,我们知道程序的编译分为四步: 1、预处理 2、编译 3、汇编 4、链接 1.1预处理 预处理功能主要包括头文件展开、宏定义、文件包含、条件编译、去注释等。 所谓的头文件展开就是在预处理时候,将头文件内容拷贝至源文…

光伏行业如何起步?光伏业务管理软件是重点!

随着全球对可再生能源的需求日益增强,光伏行业作为其中的佼佼者,正迎来前所未有的发展机遇。然而,如何在这一领域成功起步,却是许多企业和个人面临的挑战。在这个过程中,光伏业务管理软件的重要性不容忽视。 光伏行业的…

蓝桥杯历年真题省赛java b组2016年第七届

一、题目 取球博弈 两个人玩取球的游戏。 一共有N个球,每人轮流取球,每次可取集合{n1,n2,n3}中的任何一个数目。 如果无法继续取球,则游戏结束。 此时,持有奇数个球的一方获胜。 如果两人都是奇数,则为平局。 假设双…

计算机丢失vcruntime140.dll解决办法分享,有效解决vcruntime140.dll丢失问题

vcruntime140.dll是一个属于 Visual C Redistributable for Visual Studio 2015 的动态链接库文件。这个文件是运行那些用 Visual Studio 2015 或相关版本开发的 C 应用程序必不可少的一部分。如果系统中缺少此文件,或文件损坏,则可能会在尝试启动相关软…

C语言函数—递归理解和练习

练习: 编写函数不允许创建临时变量,求字符串的长度。 我们看到这道题,第一个想到的是不是strlen int main() {char[] "bit";//[b][i][t][\0]//里面一共4个字符(包括结尾的、0)但是我们的strlen函数并不会计…

mysql 常用命令

1、显示锁的时间 show status like innodb_row_lock%;2、锁一行的方法 //开启 begin; //锁一行 select * from tbl_user where name 1aa1 for update;//解锁 commit;3、设置不自动提交 set autocommit 0; //自动提交 set autocommit 1;4、查看是否支持profile show vari…

洛谷P1182数列分段

题目描述 对于给定的一个长度为 N 的正整数数列 ,现要将其分成 M(M≤N)段,并要求每段连续,且每段和的最大值最小。 关于最大值最小: 例如一数列 4 2 4 5 14 2 4 5 1 要分成 33 段。 将其如下分段&#…

信雅纳网络测试的二次开发集成:XOA(Xena Open-Source Automation)开源自动化测试

目录 XOA是什么 XOA CLI XOA Python API ​XOA Python Test Suite/测试套件 XOA Converter Source Code XOA是什么 XOA(Xena Open-Source Automation)是一个开源的测试自动化框架,追求“高效、易用、灵活”的跨操作系统的开发框架。能…