欢迎您访问365答案网,请分享给你的朋友!
生活常识 学习资料

尚硅谷大数据技术之DataX—1)概述

时间:2023-06-26
1.1 什么是DataX

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

1.2 DataX的设计

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

1.3 支持的数据源

DataX目前已经有了比较全面的插件体系,主流的RDBMS数据库、NOSQL、大数据计算系统都已经接入。

1.4 框架设计

Reader:数据采集模块,负责采集数据源的数据,将数据发送给framework。Writer:数据写入模块,负责不断向framework取数据,并将数据写入到目的端。framework:用于连接reader和writer,作为两者的数据传输通道,并处理缓冲,流控,并发,数据转换等核心技术问题。

1.5 运行原理

Job:单个作业的管理节点,负责数据清理、子任务划分、TaskGroup监控管理。Task:由Job切分而来,是DataX作业的最小单元,每个Task负责一部分数据的同步工作。Schedule:将Task组成TaskGroup,单个TaskGroup的并发数量为 5 。TaskGroup:负责启动Task。

举例来说,用户提交了一个DataX作业,并且配置了 20 个并发,目的是将一个 100 张分表的mysql数据同步到odps里面。 DataX的调度决策思路是:

1 )DataXJob根据分库分表切分成了 100 个Task。
2 )根据 20 个并发,DataX计算共需要分配 4 个TaskGroup。
3 ) 4 个TaskGroup平分切分好的 100 个Task,每一个TaskGroup负责以 5 个并发共计运行 25 个Task。

1.6 与Sqoop的对比

调度系统规避:布置多个节点,来实现实际分布式。

1.7 datax执行流程 (大数据技术之DataX)


启动步骤解析:
1、解析配置,包括job.json、core.json、plugin.json三个配置
2、设置jobId到configuration当中
3、启动Engine,通过Engine.start()进入启动程序
4、设置RUNTIME_MODE configuration当中
5、通过JobContainer的start()方法启动
6、依次执行job的preHandler()、init()、prepare()、split()、schedule()、post()、postHandle()等方法。
7、init()方法涉及到根据configuration来初始化reader和writer插件,这里涉及到jar包热加载以及调用插件init()操作方法,同时设置reader和writer的configuration信息
8、prepare()方法涉及到初始化reader和writer插件的初始化,通过调用插件的prepare()方法实现,每个插件都有自己的jarLoader,通过集成URLClassloader实现而来
9、split()方法通过adjustChannelNumber()方法调整channel个数,同时执行reader和writer最细粒度的切分,需要注意的是,writer的切分结果要参照reader的切分结果,达到切分后数目相等,才能满足1:1的通道模型
10、channel的计数主要是根据byte和record的限速来实现的,在split()的函数中第一步就是计算channel的大小
11、split()方法reader插件会根据channel的值进行拆分,但是有些reader插件可能不会参考channel的值,writer插件会完全根据reader的插件1:1进行返回
12、split()方法内部的mergeReaderAndWriterTaskConfigs()负责合并reader、writer、以及transformer三者关系,生成task的配置,并且重写job.content的配置
13、schedule()方法根据split()拆分生成的task配置分配生成taskGroup对象,根据task的数量和单个taskGroup支持的task数量进行配置,两者相除就可以得出taskGroup的数量
14、schdule()内部通过AbstractScheduler的schedule()执行,继续执行startAllTaskGroup()方法创建所有的TaskGroupContainer组织相关的task,TaskGroupContainerRunner负责运行TaskGroupContainer执行分配的task。scheduler的具体实现类为ProcessInnerScheduler。
15、taskGroupContainerExecutorService启动固定的线程池用以执行TaskGroupContainerRunner对象,TaskGroupContainerRunner的run()方法调用taskGroupContainer.start()方法,针对每个channel创建一个TaskExecutor,通过taskExecutor.doStart()启动任务

Copyright © 2016-2020 www.365daan.com All Rights Reserved. 365答案网 版权所有 备案号:

部分内容来自互联网,版权归原作者所有,如有冒犯请联系我们,我们将在三个工作时内妥善处理。