devbox
盐析白兔
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

DataX 原理解析和性能优化_datax 优化

作者：盐析白兔 | 2024-02-21 15:01:37

赞

踩

datax 优化

datax简介

datax是阿里开源的用于异构数据源之间的同步工具，由于其精巧的设计和抽象，数据同步效率极高，在很多公司数据部门都有广泛的使用。本司基于datax在阿里云普通版的rds服务器上实现了通过公网，从阿里云杭州到美国西部俄勒冈aws emr集群峰值30M以上带宽的传输效率。全量传输上亿条记录、大小30G的数据，最快不到30分钟。要知道如果拉跨洋专线的话，1M带宽每个月至少需要1千大洋呢。走公网照样能达到类似的稳定性，本文通过原理设计来阐述我们是如何基于datax做到的。

datax工作原理

在讲解datax原理之前，需要明确一些概念：

  • Job: Job是DataX用以描述从一个源头到一个目的端的同步作业，是DataX数据同步的最小业务单元。比如：从一张mysql的表同步到hive的一个表的特定分区。

  • Task: Task是为最大化而把Job拆分得到的最小执行单元。比如：读一张有1024个分表的mysql分库分表的Job，拆分成1024个读Task，若干个任务并发执行。或者将一个大表按照id拆分成1024个分片，若干个分片任务并发执行。

  • TaskGroup: 描述的是一组Task集合。在同一个TaskGroupContainer执行下的Task集合称之为TaskGroup。

  • JobContainer: Job执行器，负责Job全局拆分、调度、前置语句和后置语句等工作的工作单元。

  • TaskGroupContainer: TaskGroup执行器，负责执行一组Task的工作单元。

job和task是datax两种维度的抽象，后面源码分析中还会涉及到。

datax的处理过程可描述为：

  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。

上述流程可图像化描述为：


其中Channel是连接Reader和Writer的数据交换通道，所有的数据都会经由Channel进行传输，一个channel代表一个并发传输通道，通过该通道实现传输速率控制。接下来我们通过源码的角度，在抽取其核心逻辑，以mysql到hdfs的传输为例分析其工作流程。通过分析源码将会有以下几点收获：

  • datax 工作流程

  • datax 插件机制

  • datax 同步实现

datax源码分析

datax 工作流程

  1. public class Engine {
  2.     private static final Logger LOG = LoggerFactory.getLogger(Engine.class);
  3.  
  4.     private static String RUNTIME_MODE;
  5.  
  6.     public void start(Configuration allConf) {
  7.         boolean isJob = !("taskGroup".equalsIgnoreCase(allConf.getString(CoreConstant.DATAX_CORE_CONTAINER_MODEL)));
  8.         //JobContainer会在schedule后再行进行设置和调整值
  9.         int channelNumber =0;
  10.         AbstractContainer container;
  11.         long instanceId;
  12.         int taskGroupId = -1;
  13.             if (isJob) {
  14.             allConf.set(CoreConstant.DATAX_CORE_CONTAINER_JOB_MODE, RUNTIME_MODE);
  15.             container = new JobContainer(allConf);
  16.             instanceId = allConf.getLong(
  17.                     CoreConstant.DATAX_CORE_CONTAINER_JOB_ID, 0);
  18.  
  19.         } else {
  20.             container = new TaskGroupContainer(allConf);
  21.             instanceId = allConf.getLong(
  22.                     CoreConstant.DATAX_CORE_CONTAINER_JOB_ID);
  23.             taskGroupId = allConf.getInt(
  24.                     CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_ID);
  25.             channelNumber = allConf.getInt(
  26.                     CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_CHANNEL);
  27.         }
  28.         container.start();
  29.     }

job实例运行在jobContainer容器中，它是所有任务的master，负责初始化、拆分、调度、运行、回收、监控和汇报，但它并不做实际的数据同步操作

  1. public class JobContainer extends AbstractContainer {
  2.     private static final Logger LOG = LoggerFactory
  3.             .getLogger(JobContainer.class);
  4.  
  5.     public JobContainer(Configuration configuration) {
  6.         super(configuration);
  7.     }
  8.     /**
  9.      * jobContainer主要负责的工作全部在start()里面，包括init、prepare、split、scheduler以及destroy和statistics
  10.      */
  11.     @Override
  12.     public void start() {
  13.         LOG.info("DataX jobContainer starts job.");
  14.         try{
  15.             userConf = configuration.clone();
  16.             this.init();
  17.             this.prepare();
  18.             this.totalStage = this.split();
  19.             this.schedule();
  20.         } catch (Throwable e) {
  21.             Communication communication = super.getContainerCommunicator().collect();
  22.             // 汇报前的状态，不需要手动进行设置
  23.             // communication.setState(State.FAILED);
  24.             communication.setThrowable(e);
  25.             communication.setTimestamp(this.endTimeStamp);
  26.  
  27.             Communication tempComm = new Communication();
  28.             tempComm.setTimestamp(this.startTransferTimeStamp);
  29.  
  30.             Communication reportCommunication = CommunicationTool.getReportCommunication(communication, tempComm, this.totalStage);
  31.             super.getContainerCommunicator().report(reportCommunication);
  32.  
  33.             throw DataXException.asDataXException(
  34.                     FrameworkErrorCode.RUNTIME_ERROR, e);
  35.         }
  36.     }
  37.  
  38.     /**
  39.      * reader和writer的初始化
  40.     */
  41.     private void init() {
  42.         Thread.currentThread().setName("job-" + this.jobId);
  43.  
  44.         JobPluginCollector jobPluginCollector = new DefaultJobPluginCollector(
  45.                 this.getContainerCommunicator());
  46.         //必须先Reader ，后Writer
  47.         this.jobReader = this.initJobReader(jobPluginCollector);
  48.         this.jobWriter = this.initJobWriter(jobPluginCollector);
  49.     }
  50.  
  51.     /**
  52.      *schedule首先完成的工作是把上一步reader和writer split的结果整合到具体taskGroupContainer中,
  53.      * 同时不同的执行模式调用不同的调度策略，将所有任务调度起来
  54.      */
  55.     private void schedule() {
  56.         /**
  57.          * 这里的全局speed和每个channel的速度设置为B/s
  58.          */
  59.         int channelsPerTaskGroup = this.configuration.getInt(
  60.                 CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_CHANNEL, 5);
  61.         int taskNumber = this.configuration.getList(
  62.                 CoreConstant.DATAX_JOB_CONTENT).size();
  63.  
  64.         this.needChannelNumber = Math.min(this.needChannelNumber, taskNumber);
  65.         /**
  66.          * 通过获取配置信息得到每个taskGroup需要运行哪些tasks任务。
  67.          会考虑 task 中对资源负载作的 load 标识进行更均衡的作业分配操作。
  68.          */
  69.         List<Configuration> taskGroupConfigs = JobAssignUtil.assignFairly(this.configuration,
  70.                 this.needChannelNumber, channelsPerTaskGroup);
  71.                         LOG.info("Scheduler starts [{}] taskGroups.", taskGroupConfigs.size());
  72.  
  73.         AbstractScheduler scheduler;
  74.         try {
  75.             scheduler = initStandaloneScheduler(this.configuration);
  76.  
  77.             this.startTransferTimeStamp = System.currentTimeMillis();
  78.  
  79.             scheduler.schedule(taskGroupConfigs);
  80.  
  81.             this.endTransferTimeStamp = System.currentTimeMillis();
  82.         } catch (Exception e) {
  83.             LOG.error("运行scheduler出错.");
  84.             this.endTransferTimeStamp = System.currentTimeMillis();
  85.             throw DataXException.asDataXException(
  86.                     FrameworkErrorCode.RUNTIME_ERROR, e);
  87.         }
  88.     }
  89.  
  90.     private AbstractScheduler initStandaloneScheduler(Configuration configuration) {
  91.         AbstractContainerCommunicator containerCommunicator = new StandAloneJobContainerCommunicator(configuration);
  92.         super.setContainerCommunicator(containerCommunicator);
  93.  
  94.         return new StandAloneScheduler(containerCommunicator);
  95.     }
  96. }    
  97.  
  98. public abstract class AbstractScheduler {
  99.     private static final Logger LOG = LoggerFactory
  100.             .getLogger(AbstractScheduler.class);
  101.     public void schedule(List<Configuration> configurations) {
  102.         /**
  103.          * 给 taskGroupContainer 的 Communication 注册
  104.          */
  105.         this.containerCommunicator.registerCommunication(configurations);
  106.         int totalTasks = calculateTaskCount(configurations);
  107.         startAllTaskGroup(configurations);
  108.         try {
  109.             while (true) {
  110.                 Communication nowJobContainerCommunication = this.containerCommunicator.collect();
  111.                 //汇报周期
  112.                 long now = System.currentTimeMillis();
  113.                 if (now - lastReportTimeStamp > jobReportIntervalInMillSec) {
  114.                     Communication reportCommunication = CommunicationTool
  115.                             .getReportCommunication(nowJobContainerCommunication, lastJobContainerCommunication, totalTasks);
  116.  
  117.                     this.containerCommunicator.report(reportCommunication);
  118.                      if (nowJobContainerCommunication.getState() == State.SUCCEEDED) {
  119.                     LOG.info("Scheduler accomplished all tasks.");
  120.                     break;
  121.                 }
  122.                 if (nowJobContainerCommunication.getState() == State.FAILED) {
  123.                     dealFailedStat(this.containerCommunicator, nowJobContainerCommunication.getThrowable());
  124.                 }
  125.  
  126.                 Thread.sleep(jobSleepIntervalInMillSec);
  127.             }
  128.         } catch (InterruptedException e) {
  129.             // 以 failed 状态退出
  130.             LOG.error("捕获到InterruptedException异常!", e);
  131.  
  132.             throw DataXException.asDataXException(
  133.                     FrameworkErrorCode.RUNTIME_ERROR, e);
  134.         }
  135.     }
  136.  
  137.     @Override
  138.     public void startAllTaskGroup(List<Configuration> configurations) {
  139.         this.taskGroupContainerExecutorService = Executors
  140.                 .newFixedThreadPool(configurations.size());
  141.  
  142.         for (Configuration taskGroupConfiguration : configurations) {
  143.             TaskGroupContainerRunner taskGroupContainerRunner = newTaskGroupContainerRunner(taskGroupConfiguration);
  144.             this.taskGroupContainerExecutorService.execute(taskGroupContainerRunner);
  145.         }
  146.  
  147.         this.taskGroupContainerExecutorService.shutdown();
  148.     }
  149.  
  150.  
  151.     @Override
  152.     public void dealFailedStat(AbstractContainerCommunicator frameworkCollector, Throwable throwable) {
  153.         this.taskGroupContainerExecutorService.shutdownNow();
  154.     }
  155.  
  156. }
  157.  
  158. public class TaskGroupContainer extends AbstractContainer {
  159.     private static final Logger LOG = LoggerFactory
  160.             .getLogger(TaskGroupContainer.class);
  161.             @Override
  162.     public void start() {
  163.         try {
  164.             while (true) {
  165.                 //1.判断task状态
  166.                 boolean failedOrKilled = false;
  167.                 Map<Integer, Communication> communicationMap = containerCommunicator.getCommunicationMap();
  168.                 for(Map.Entry<Integer, Communication> entry :         communicationMap.entrySet()){
  169.                     Integer taskId = entry.getKey();
  170.                     Communication taskCommunication = entry.getValue();
  171.                     if(!taskCommunication.isFinished()){
  172.                         continue;
  173.                     }
  174.                     TaskExecutor taskExecutor = removeTask(runTasks, taskId);
  175.                     if(taskCommunication.getState() == State.FAILED){
  176.                         failedOrKilled = true;
  177.                             break;
  178.                     }
  179.                     else if(taskCommunication.getState() == State.SUCCEEDED){
  180.                                                Long taskStartTime = taskStartTimeMap.get(taskId);
  181.                         if(taskStartTime != null){
  182.                             Long usedTime = System.currentTimeMillis() - taskStartTime;
  183.                             LOG.info("taskGroup[{}] taskId[{}] is successed, used[{}]ms",
  184.                                     this.taskGroupId, taskId, usedTime);
  185.                             //usedTime*1000*1000 
  186.                             taskStartTimeMap.remove(taskId);
  187.                             taskConfigMap.remove(taskId);
  188.                         }
  189.                     }
  190.                 }
  191.                 // 2.发现该taskGroup下taskExecutor的总状态失败则汇报错误
  192.                 if (failedOrKilled) {
  193.                     lastTaskGroupContainerCommunication = reportTaskGroupCommunication(
  194.                             lastTaskGroupContainerCommunication, taskCountInThisTaskGroup);
  195.  
  196.                     throw DataXException.asDataXException(
  197.                             FrameworkErrorCode.PLUGIN_RUNTIME_ERROR, lastTaskGroupContainerCommunication.getThrowable());
  198.                 }
  199.                 //3.有任务未执行，且正在运行的任务数小于最大通道限制
  200.                 Iterator<Configuration> iterator = taskQueue.iterator();
  201.                 while(iterator.hasNext() && runTasks.size() < channelNumber){
  202.                     Configuration taskConfig = iterator.next();
  203.                     Integer taskId = taskConfig.getInt(CoreConstant.TASK_ID);
  204.                     Configuration taskConfigForRun =taskConfig.clone()
  205.                     TaskExecutor taskExecutor = new TaskExecutor(taskConfigForRun);
  206.                     taskStartTimeMap.put(taskId, System.currentTimeMillis());
  207.                     taskExecutor.doStart();
  208.                     terator.remove();
  209.                     runTasks.add(taskExecutor);
  210.                     LOG.info("taskGroup[{}] taskId[{}] is started",
  211.                             this.taskGroupId, taskId);
  212.             }
  213.             //4.任务列表为空，executor已结束, 搜集状态为success--->成功
  214.             if (taskQueue.isEmpty() && isAllTaskDone(runTasks) && containerCommunicator.collectState() == State.SUCCEEDED) {
  215.                 // 成功的情况下，也需要汇报一次。否则在任务结束非常快的情况下，采集的信息将会不准确
  216.                 lastTaskGroupContainerCommunication = reportTaskGroupCommunication(
  217.                         lastTaskGroupContainerCommunication, taskCountInThisTaskGroup);
  218.  
  219.                 LOG.info("taskGroup[{}] completed it's tasks.", this.taskGroupId);
  220.                 break;
  221.             }
  222.         } catch (Throwable e) {
  223.             Communication nowTaskGroupContainerCommunication = this.containerCommunicator.collect();
  224.  
  225.             if (nowTaskGroupContainerCommunication.getThrowable() == null) {
  226.                 nowTaskGroupContainerCommunication.setThrowable(e);
  227.             }
  228.             nowTaskGroupContainerCommunication.setState(State.FAILED);
  229.             this.containerCommunicator.report(nowTaskGroupContainerCommunication);
  230.  
  231.             throw DataXException.asDataXException(
  232.                     FrameworkErrorCode.RUNTIME_ERROR, e);
  233.         }
  234.     }
  235.  
  236. }
  237. /**
  238.  * TaskExecutor是一个完整task的执行器
  239.  * 其中包括1：1的reader和writer
  240.  */
  241. class TaskExecutor {
  242.  
  243.     private Thread readerThread;
  244.  
  245.     private Thread writerThread;
  246.  
  247.     private ReaderRunner readerRunner;
  248.  
  249.     private WriterRunner writerRunner;
  250.  
  251.     public TaskExecutor(Configuration taskConf, int attemptCount) {
  252.         writerRunner = (WriterRunner) generateRunner(PluginType.WRITER);
  253.         //生成writerThread
  254.         this.writerThread = new Thread(writerRunner,
  255.                     String.format("%d-%d-%d-writer",
  256.                             jobId, taskGroupId, this.taskId));
  257.         //生成readerThread
  258.         readerRunner = (ReaderRunner) generateRunner(PluginType.READER,transformerInfoExecs);
  259.         this.readerThread = new Thread(readerRunner,
  260.                     String.format("%d-%d-%d-reader",
  261.                             jobId, taskGroupId, this.taskId));
  262.     }
  263.  
  264.     public void doStart() {
  265.         this.writerThread.start();
  266.         // reader没有起来，writer不可能结束
  267.         if (!this.writerThread.isAlive() || this.taskCommunication.getState() == State.FAILED) {
  268.             throw DataXException.asDataXException(
  269.                     FrameworkErrorCode.RUNTIME_ERROR,
  270.                     this.taskCommunication.getThrowable());
  271.         }
  272.  
  273.         this.readerThread.start();
  274.  
  275.         // 这里reader可能很快结束
  276.         if (!this.readerThread.isAlive() && this.taskCommunication.getState() == State.FAILED) {
  277.             // 这里有可能出现Reader线上启动即挂情况 对于这类情况 需要立刻抛出异常
  278.             throw DataXException.asDataXException(
  279.                     FrameworkErrorCode.RUNTIME_ERROR,
  280.                     this.taskCommunication.getThrowable());
  281.         }
  282.     }
  283. }

从上面总体流程中可以看到JobContainer通过线程池调度起所有的TaskGroupContainer，然后轮训TaskGroupContainer的运行状态。每个TaskGroupContainer则是根据分配的chanel并发数量依次执行分配的Task实例。

插件机制

在工作流程中的init步骤，我们看到的jobReader和jobWriter的实现就是通过插件动态生成的。jobReader和jobWriter就对应datax中的Job概念模型。而在TaskExecutor中初始化的readerRunner和writerRunner对应的是Task模型。通过插件datax插件机制支持了数十种不同的数据源之间的读写同步，同时也很方便的支持新的数据源接入。

Job初始化过程

  1. public class JobContainer extends AbstractContainer {
  2.  
  3.     //reader job的初始化，返回Reader.Job
  4.     private Reader.Job initJobReader(
  5.             JobPluginCollector jobPluginCollector) {
  6.         this.readerPluginName = this.configuration.getString(
  7.                 CoreConstant.DATAX_JOB_CONTENT_READER_NAME);
  8.  
  9.         Reader.Job jobReader = (Reader.Job) LoadUtil.loadJobPlugin(
  10.                 PluginType.READER, this.readerPluginName);
  11.  
  12.         // 设置reader的jobConfig
  13.         jobReader.setPluginJobConf(this.configuration.getConfiguration(
  14.                 CoreConstant.DATAX_JOB_CONTENT_READER_PARAMETER));
  15.  
  16.         // 设置reader的readerConfig
  17.         jobReader.setPeerPluginJobConf(this.configuration.getConfiguration(
  18.                 CoreConstant.DATAX_JOB_CONTENT_WRITER_PARAMETER));
  19.  
  20.         jobReader.setJobPluginCollector(jobPluginCollector);
  21.         jobReader.init();
  22.  
  23.         classLoaderSwapper.restoreCurrentThreadClassLoader();
  24.         return jobReader;
  25.     }
  26. }

插件加载器，大体上分reader、transformer（还未实现）和writer三中插件类型，
reader和writer在执行时又可能出现Job和Task两种运行时（加载的类不同）

  1. public class LoadUtil {
  2.     //加载JobPlugin，reader、writer都可能要加载
  3.     public static AbstractJobPlugin loadJobPlugin(PluginType pluginType,
  4.                                                   String pluginName) {
  5.         Class<? extends AbstractPlugin> clazz = LoadUtil.loadPluginClass(
  6.                 pluginType, pluginName, ContainerType.Job);
  7.  
  8.         try {
  9.             AbstractJobPlugin jobPlugin = (AbstractJobPlugin) clazz
  10.                     .newInstance();
  11.             jobPlugin.setPluginConf(getPluginConf(pluginType, pluginName));
  12.             return jobPlugin;
  13.         } catch (Exception e) {
  14.             throw DataXException.asDataXException(
  15.                     FrameworkErrorCode.RUNTIME_ERROR,
  16.                     String.format("DataX找到plugin[%s]的Job配置.",
  17.                             pluginName), e);
  18.         }
  19.     }
  20.  
  21.     //反射出具体plugin实例
  22.     private static synchronized Class<? extends AbstractPlugin> loadPluginClass(
  23.             PluginType pluginType, String pluginName,
  24.             ContainerType pluginRunType) {
  25.         Configuration pluginConf = getPluginConf(pluginType, pluginName);
  26.         JarLoader jarLoader = LoadUtil.getJarLoader(pluginType, pluginName);
  27.         try {
  28.             return (Class<? extends AbstractPlugin>) jarLoader
  29.                     .loadClass(pluginConf.getString("class") + "$"
  30.                             + pluginRunType.value());
  31.         } catch (Exception e) {
  32.             throw DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR, e);
  33.         }
  34.     }
  35.  
  36.     public static synchronized JarLoader getJarLoader(PluginType pluginType,
  37.                                                       String pluginName) {
  38.         Configuration pluginConf = getPluginConf(pluginType, pluginName);
  39.  
  40.         JarLoader jarLoader = jarLoaderCenter.get(generatePluginKey(pluginType,
  41.                 pluginName));
  42.         if (null == jarLoader) {
  43.             String pluginPath = pluginConf.getString("path");
  44.             if (StringUtils.isBlank(pluginPath)) {
  45.                 throw DataXException.asDataXException(
  46.                         FrameworkErrorCode.RUNTIME_ERROR,
  47.                         String.format(
  48.                                 "%s插件[%s]路径非法!",
  49.                                 pluginType, pluginName));
  50.             }
  51.             jarLoader = new JarLoader(new String[]{pluginPath});
  52.             jarLoaderCenter.put(generatePluginKey(pluginType, pluginName),
  53.                     jarLoader);
  54.         }
  55.  
  56.         return jarLoader;
  57.     }
  58.  
  59. }
  60.  
  61. //提供Jar隔离的加载机制，会把传入的路径、及其子路径、以及路径中的jar文件加入到class path。
  62. public class JarLoader extends URLClassLoader {
  63.     public JarLoader(String[] paths) {
  64.         this(paths, JarLoader.class.getClassLoader());
  65.     }
  66.  
  67.     public JarLoader(String[] paths, ClassLoader parent) {
  68.         super(getURLs(paths), parent);
  69.     }
  70.  
  71.     private static URL[] getURLs(String[] paths) {
  72.         Validate.isTrue(null != paths && 0 != paths.length,
  73.                 "jar包路径不能为空.");
  74.  
  75.         List<String> dirs = new ArrayList<String>();
  76.         for (String path : paths) {
  77.             dirs.add(path);
  78.             JarLoader.collectDirs(path, dirs);
  79.         }
  80.  
  81.         List<URL> urls = new ArrayList<URL>();
  82.         for (String path : dirs) {
  83.             urls.addAll(doGetURLs(path));
  84.         }
  85.  
  86.         return urls.toArray(new URL[0]);
  87.     }
  88.  
  89.     private static void collectDirs(String path, List<String> collector) {
  90.         if (null == path || StringUtils.isBlank(path)) {
  91.             return;
  92.         }
  93.  
  94.         File current = new File(path);
  95.         if (!current.exists() || !current.isDirectory()) {
  96.             return;
  97.         }
  98.  
  99.         for (File child : current.listFiles()) {
  100.             if (!child.isDirectory()) {
  101.                 continue;
  102.             }
  103.  
  104.             collector.add(child.getAbsolutePath());
  105.             collectDirs(child.getAbsolutePath(), collector);
  106.         }
  107.     }
  108. }

Task 初始化过程

  1. class TaskExecutor {
  2. private AbstractRunner generateRunner(PluginType pluginType) {
  3.             return generateRunner(pluginType, null);
  4.         }
  5.  
  6.         private AbstractRunner generateRunner(PluginType pluginType, List<TransformerExecution> transformerInfoExecs) {
  7.             AbstractRunner newRunner = null;
  8.             TaskPluginCollector pluginCollector;
  9.  
  10.             switch (pluginType) {
  11.                 case READER:
  12.                     newRunner = LoadUtil.loadPluginRunner(pluginType,
  13.                             this.taskConfig.getString(CoreConstant.JOB_READER_NAME));
  14.                     newRunner.setJobConf(this.taskConfig.getConfiguration(
  15.                             CoreConstant.JOB_READER_PARAMETER));
  16.  
  17.                     pluginCollector = ClassUtil.instantiate(
  18.                             taskCollectorClass, AbstractTaskPluginCollector.class,
  19.                             configuration, this.taskCommunication,
  20.                             PluginType.READER);
  21.  
  22.                     RecordSender recordSender;
  23.                     if (transformerInfoExecs != null && transformerInfoExecs.size() > 0) {
  24.                         recordSender = new BufferedRecordTransformerExchanger(taskGroupId, this.taskId, this.channel,this.taskCommunication ,pluginCollector, transformerInfoExecs);
  25.                     } else {
  26.                         recordSender = new BufferedRecordExchanger(this.channel, pluginCollector);
  27.                     }
  28.  
  29.                     ((ReaderRunner) newRunner).setRecordSender(recordSender);
  30.  
  31.                     /**
  32.                      * 设置taskPlugin的collector，用来处理脏数据和job/task通信
  33.                      */
  34.                     newRunner.setTaskPluginCollector(pluginCollector);
  35.                     break;
  36.                 case WRITER:
  37.                     newRunner = LoadUtil.loadPluginRunner(pluginType,
  38.                             this.taskConfig.getString(CoreConstant.JOB_WRITER_NAME));
  39.                     newRunner.setJobConf(this.taskConfig
  40.                             .getConfiguration(CoreConstant.JOB_WRITER_PARAMETER));
  41.  
  42.                     pluginCollector = ClassUtil.instantiate(
  43.                             taskCollectorClass, AbstractTaskPluginCollector.class,
  44.                             configuration, this.taskCommunication,
  45.                             PluginType.WRITER);
  46.                     ((WriterRunner) newRunner).setRecordReceiver(new BufferedRecordExchanger(
  47.                             this.channel, pluginCollector));
  48.                     /**
  49.                      * 设置taskPlugin的collector，用来处理脏数据和job/task通信
  50.                      */
  51.                     newRunner.setTaskPluginCollector(pluginCollector);
  52.                     break;
  53.                 default:
  54.                     throw DataXException.asDataXException(FrameworkErrorCode.ARGUMENT_ERROR, "Cant generateRunner for:" + pluginType);
  55.             }
  56.  
  57.             newRunner.setTaskGroupId(taskGroupId);
  58.             newRunner.setTaskId(this.taskId);
  59.             newRunner.setRunnerCommunication(this.taskCommunication);
  60.  
  61.             return newRunner;
  62.         }
  63. }
  64.  
  65. public class LoadUtil {
  66.   /**
  67.      * 根据插件类型、名字和执行时taskGroupId加载对应运行器
  68.      *
  69.      * @param pluginType
  70.      * @param pluginName
  71.      * @return
  72.      */
  73.     public static AbstractRunner loadPluginRunner(PluginType pluginType, String pluginName) {
  74.         AbstractTaskPlugin taskPlugin = LoadUtil.loadTaskPlugin(pluginType,
  75.                 pluginName);
  76.  
  77.         switch (pluginType) {
  78.             case READER:
  79.                 return new ReaderRunner(taskPlugin);
  80.             case WRITER:
  81.                 return new WriterRunner(taskPlugin);
  82.             default:
  83.                 throw DataXException.asDataXException(
  84.                         FrameworkErrorCode.RUNTIME_ERROR,
  85.                         String.format("插件[%s]的类型必须是[reader]或[writer]!",
  86.                                 pluginName));
  87.         }
  88.     }
  89. }

同步实现

这部分就是经过split后的具体的Task的执行逻辑。Task的划分逻辑如下：

  1. public class JobContainer extends AbstractContainer {
  2.     private static final Logger LOG = LoggerFactory
  3.             .getLogger(JobContainer.class);
  4.     /**
  5.      * 执行reader和writer最细粒度的切分，需要注意的是，writer的切分结果要参照reader的切分结果，
  6.      * 达到切分后数目相等，才能满足1：1的通道模型，所以这里可以将reader和writer的配置整合到一起，
  7.      * 然后，为避免顺序给读写端带来长尾影响，将整合的结果shuffler掉
  8.      */
  9.     private int split() {
  10.         this.adjustChannelNumber();
  11.  
  12.         if (this.needChannelNumber <= 0) {
  13.             this.needChannelNumber = 1;
  14.         }
  15.  
  16.         List<Configuration> readerTaskConfigs = this
  17.                 .doReaderSplit(this.needChannelNumber);
  18.         int taskNumber = readerTaskConfigs.size();
  19.         List<Configuration> writerTaskConfigs = this
  20.                 .doWriterSplit(taskNumber);
  21.  
  22.         List<Configuration> transformerList = this.configuration.getListConfiguration(CoreConstant.DATAX_JOB_CONTENT_TRANSFORMER);
  23.  
  24.         LOG.debug("transformer configuration: "+ JSON.toJSONString(transformerList));
  25.         /**
  26.          * 输入是reader和writer的parameter list，输出是content下面元素的list
  27.          */
  28.         List<Configuration> contentConfig = mergeReaderAndWriterTaskConfigs(
  29.                 readerTaskConfigs, writerTaskConfigs, transformerList);
  30.  
  31.  
  32.         LOG.debug("contentConfig configuration: "+ JSON.toJSONString(contentConfig));
  33.  
  34.         this.configuration.set(CoreConstant.DATAX_JOB_CONTENT, contentConfig);
  35.  
  36.         return contentConfig.size();
  37.     }
  38.  
  39. }

 每个Task都执行相同的逻辑和流程，下面以读mysql和写hdfs为例，展示其读写过程。

  1.  //单个slice的reader执行调用
  2. public class ReaderRunner extends AbstractRunner implements Runnable {
  3.    @Override
  4.     public void run() {
  5.         Reader.Task taskReader = (Reader.Task) this.getPlugin();
  6.         taskReader.init();
  7.         taskReader.prepare();
  8.         taskReader.startRead(recordSender);
  9.         recordSender.terminate();
  10.     }
  11. }
  12.  
  13. public class MysqlReader extends Reader {
  14.         @Override
  15.         public void startRead(RecordSender recordSender) {
  16.             int fetchSize = this.readerSliceConfig.getInt(Constant.FETCH_SIZE);
  17.  
  18.             this.commonRdbmsReaderTask.startRead(this.readerSliceConfig, recordSender,
  19.                     super.getTaskPluginCollector(), fetchSize);
  20.         }
  21. }
  22.  
  23. public class CommonRdbmsReader {
  24.         public static class Task {
  25.         private static final Logger LOG = LoggerFactory
  26.                 .getLogger(Task.class);
  27.         public void startRead(Configuration readerSliceConfig,
  28.                               RecordSender recordSender,
  29.                               TaskPluginCollector taskPluginCollector, int fetchSize) {
  30.             String querySql = readerSliceConfig.getString(Key.QUERY_SQL);
  31.             String table = readerSliceConfig.getString(Key.TABLE);
  32.  
  33.             PerfTrace.getInstance().addTaskDetails(taskId, table + "," + basicMsg);
  34.  
  35.             LOG.info("Begin to read record by Sql: [{}\n] {}.",
  36.                     querySql, basicMsg);
  37.  
  38.             Connection conn = DBUtil.getConnection(this.dataBaseType, jdbcUrl,
  39.                     username, password);
  40.             int columnNumber = 0;
  41.             ResultSet rs = null;
  42.             try {
  43.                 rs = DBUtil.query(conn, querySql, fetchSize);
  44.                 while (rs.next()) {
  45.                     //将数据记录放入channel通道，writer从中获取写数据
  46.                     this.transportOneRecord(recordSender, rs,
  47.                             metaData, columnNumber, mandatoryEncoding, taskPluginCollector);
  48.                 }
  49.             }catch (Exception e) {
  50.                 throw RdbmsException.asQueryException(this.dataBaseType, e, querySql, table, username);
  51.             } finally {
  52.                 DBUtil.closeDBResources(null, conn);
  53.             }
  54.         }
  55.     }
  56. }

 //单个slice的writer执行调用

  1. public class WriterRunner extends AbstractRunner implements Runnable {
  2.     @Override
  3.     public void run() {
  4.         Writer.Task taskWriter = (Writer.Task) this.getPlugin();
  5.         taskWriter.init();
  6.         taskWriter.prepare();
  7.         taskWriter.startWrite(recordReceiver);
  8.     }
  9. }
  10.  
  11. public class HdfsWriter extends Writer {
  12.     public static class Task extends Writer.Task {
  13.         private static final Logger LOG = LoggerFactory.getLogger(Task.class);
  14.  
  15.         @Override
  16.         public void startWrite(RecordReceiver lineReceiver) {
  17.             LOG.info("begin do write...");
  18.             LOG.info(String.format("write to file : [%s]", this.fileName));
  19.             if(fileType.equalsIgnoreCase("TEXT")){
  20.                 //写TEXT FILE
  21.                 hdfsHelper.textFileStartWrite(lineReceiver,this.writerSliceConfig, this.fileName,
  22.                         this.getTaskPluginCollector());
  23.             }else if(fileType.equalsIgnoreCase("ORC")){
  24.                 //写ORC FILE
  25.                 hdfsHelper.orcFileStartWrite(lineReceiver,this.writerSliceConfig, this.fileName,
  26.                         this.getTaskPluginCollector());
  27.             }
  28.             LOG.info("end do write");
  29.         }
  30.     }
  31. }
  32.  
  33. public  class HdfsHelper {
  34.     public void textFileStartWrite(RecordReceiver lineReceiver, Configuration config, String fileName,TaskPluginCollector taskPluginCollector){
  35.         try {
  36.             RecordWriter writer = outFormat.getRecordWriter(fileSystem, conf, outputPath.toString(), Reporter.NULL);
  37.             Record record = null;
  38.             while ((record = lineReceiver.getFromReader()) != null) {
  39.                 MutablePair<Text, Boolean> transportResult = transportOneRecord(record, fieldDelimiter, columns, taskPluginCollector);
  40.                 if (!transportResult.getRight()) {
  41.                     writer.write(NullWritable.get(),transportResult.getLeft());
  42.                 }
  43.             }
  44.             writer.close(Reporter.NULL);
  45.         } catch (Exception e) {
  46.             String message = String.format("写文件文件[%s]时发生IO异常,请检查您的网络是否正常！", fileName);
  47.             LOG.error(message);
  48.             Path path = new Path(fileName);
  49.             deleteDir(path.getParent());
  50.             throw DataXException.asDataXException(HdfsWriterErrorCode.Write_FILE_IO_ERROR, e);
  51.         }
  52.     }
  53. }

reader和writer通过BufferedRecordExchanger建立联系，在其内部实现了基于ArrayBlockingQueue的MemoryChannel。

  1. public class BufferedRecordExchanger implements RecordSender, RecordReceiver {
  2.     @Override
  3.     public void sendToWriter(Record record) {
  4.         if(shutdown){
  5.             throw DataXException.asDataXException(CommonErrorCode.SHUT_DOWN_TASK, "");
  6.         }
  7.  
  8.         Validate.notNull(record, "record不能为空.");
  9.  
  10.         if (record.getMemorySize() > this.byteCapacity) {
  11.             this.pluginCollector.collectDirtyRecord(record, new Exception(String.format("单条记录超过大小限制，当前限制为:%s", this.byteCapacity)));
  12.             return;
  13.         }
  14.  
  15.         boolean isFull = (this.bufferIndex >= this.bufferSize || this.memoryBytes.get() + record.getMemorySize() > this.byteCapacity);
  16.         if (isFull) {
  17.             flush();
  18.         }
  19.  
  20.         this.buffer.add(record);
  21.         this.bufferIndex++;
  22.         memoryBytes.addAndGet(record.getMemorySize());
  23.     }
  24.  
  25.     @Override
  26.     public void flush() {
  27.         if(shutdown){
  28.             throw DataXException.asDataXException(CommonErrorCode.SHUT_DOWN_TASK, "");
  29.         }
  30.         this.channel.pushAll(this.buffer);
  31.         this.buffer.clear();
  32.         this.bufferIndex = 0;
  33.         this.memoryBytes.set(0);
  34.     }
  35.  
  36.     @Override
  37.     public Record getFromReader() {
  38.         if(shutdown){
  39.             throw DataXException.asDataXException(CommonErrorCode.SHUT_DOWN_TASK, "");
  40.         }
  41.         boolean isEmpty = (this.bufferIndex >= this.buffer.size());
  42.         if (isEmpty) {
  43.             receive();
  44.         }
  45.  
  46.         Record record = this.buffer.get(this.bufferIndex++);
  47.         if (record instanceof TerminateRecord) {
  48.             record = null;
  49.         }
  50.         return record;
  51.     }

datax性能优化

通过datax原理和实现的理解，自然可以知道如何提升datax的同步效率。以mysql同步hdfs为例，自然最直接的方式就是提高mysql和hdfs的硬件性能如cpu、内存、IOPS、网络带宽等。当硬件资源受限的情况下，可以有如下几种办法：

  1. 将不同的集群划分到同一个网络或者区域内，减少跨网络的不稳定性，如将阿里云集群迁移到amazon集群，或者同一个amazon集群中不同区域划分到同一个子网络内。

  2. 对数据库按照主键划分。datax对单个表默认一个通道，如果指定拆分主键，将会大大提升同步并发数和吞吐量。

  3. 在cpu、内存以及mysql负载满足的情况下，提升通道并发数。通道并发数意味着更多的内存开销，jvm调优是重中之重。

  4. 当无法提升通道数量时，而且每个拆分依然很大的时候，可以考虑对每个拆分再次拆分。

  5. 设定合适的参数，如mysql超时等。

总结

本文通过原理介绍和源码分析，逐步理清datax的工作流程和实现原理，并结合实际经验给出几点优化建议。然而在实际中涉及到更多的分库分表和特大量级的表，数据库的承载压力也是一大考虑因素，否则遭到dba的吊打肯定会在所难免。尤其是我们涉及到跨大洋数据同步，网络的稳定性也是一大挑战，此时基于增量同步方案或许是更好的选择。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/盐析白兔/article/detail/124554?site
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号