DataX3.0原理分析

介绍

阿里开源的一个异构数据源离线同步工具。包括：关系型数据库（Mysql，Oracel等），HDFS、Hive、ODPS、HBase、FTP等各种异构数据源。

在阿里集团广泛使用，自2015年开始就稳定运行，每天完成8w作业，日传输数据量300TB。

框架设计

Reader：数据采集。

Writer：数据写入。

Framework：连接reader和writer，作为两者的通道，并处理数据的缓冲，流控，并发和数据转换等功能。

架构

流程介绍：

1. DataX完成单个数据同步的作业，我们称之为Job，DataX接受到一个Job之后，将启动一个进程来完成整个作业同步过程。DataX Job模块是单个作业的中枢管理节点，承担了数据清理、子任务切分(将单一作业计算转化为多个子Task)、TaskGroup管理等功能。

2. DataXJob启动后，会根据不同的源端切分策略，将Job切分成多个小的Task(子任务)，以便于并发执行。Task便是DataX作业的最小单元，每一个Task都会负责一部分数据的同步工作。

3. 切分多个Task之后，DataX Job会调用Scheduler模块，根据配置的并发数据量，将拆分成的Task重新组合，组装成TaskGroup(任务组)。每一个TaskGroup负责以一定的并发运行完毕分配好的所有Task，默认单个任务组的并发数量为5。

4. 每一个Task都由TaskGroup负责启动，Task启动后，会固定启动Reader—>Channel—>Writer的线程来完成任务同步工作。

5. DataX作业运行起来之后， Job监控并等待多个TaskGroup模块任务完成，等待所有TaskGroup任务完成后Job成功退出。否则，异常退出，进程退出值非0。

UML

插件架构

注：如果扩展数据源，则只需要实现reader和writer接口

容器架构

流程详情

Engine

关键代码：

public void start(Configuration allConf)
    AbstractContainer container;
    //根据配置：core.container.model
    if (isJob) {
        //启动job容器
        container = new JobContainer(allConf);
    } else {
        //启动task容器
        container = new TaskGroupContainer(allConf);
    }
    container.start();
}

JobContainer

关键代码：

public class JobContainer extends AbstractContainer
    public void start()
        //拷贝一份新的配置，保证线程安全
        userConf = configuration.clone();
        //前置处理
        this.preHandle();
        //初始化插件，先Reader，再初始化Writer
        this.init();
        //根据插件jar路径设置reader和writer类加载器，执行插件自定义逻辑。
        this.prepare();
        //任务切分
        this.split();
        //将Task拆分成taskGroup。taskGroup会启动线程池运行TaskGroupContainer
        this.schedule();
        //根据插件jar路径设置reader和writer类加载器，执行插件自定义逻辑。
        this.post();
        //后置处理
        this.postHandle();
        //调用外部hook
        this.invokeHooks();
    }
}

1. 前置处理和后置处理使用的类加载器是根据参数配置的。

   job.preHandler.pluginType
   job.preHandler.pluginName

2. prepare和post的类加载器由：writer/reader和

   job.content[0].writer.name
   job.content[0].reader.name

   来决定。

3. 任务的切分有3种配置：

   1. table模式：eachTableShouldSplittedNumber=1.0 * adviceNumber / tableNumber。adviceNumber：这里是channel数量大小, 即datax并发task数量) ； tableNumber:表数量)adviceNumber

   2. sql语句查询模式：根据sql的数据量进行task数量的切分。TaskGroupContainer

TaskGroupContainer

关键代码：

public class TaskGroupContainer extends AbstractContainer
    public void start(){
        //根据配置获取channel数量，core.container.taskGroup.channel
        int channelNumber = this.configuration.getInt(
					CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_CHANNEL);
	//生成TaskExecutor：task执行器
	TaskExecutor taskExecutor = new TaskExecutor(taskConfigForRun, attemptCount);
	taskExecutor.doStart();
    }
}

TaskExecutor

task执行器

生成读写线程

//写线程
this.writerThread = new Thread(writerRunner,
	String.format("%d-%d-%d-writer",
	jobId, taskGroupId, this.taskId));
//读线程
this.readerThread = new Thread(readerRunner,
String.format("%d-%d-%d-reader",
jobId, taskGroupId, this.taskId));			

生成读写数据载体

private AbstractRunner generateRunner(){
    RecordSender recordSender;
    //数据载体类型：
    //BufferedRecordExchanger
    //BufferedRecordTransformerExchanger
    //RecordExchanger
    //TransformerExchanger    
}

Channel

数据载体是通过Channel作为管道来进行的数据传输。

//内存channel实现
public class MemoryChannel extends Channel
    //阻塞队列
    private ArrayBlockingQueue<Record> queue = null；
}