datax(9):Job和TaskGroup的通讯机制

datax(9):Job和TaskGroup的通讯机制先后看完了TaskGroupContainer和JobContainer,梳理下他们的关系与职责;一,各自职责JobContainer:Job执行器,负责Job全局拆分、调度、前置语句和后置语句等工作的工作单元。类似Yarn中的JobTrackerTaskGroupContainer:TaskGroup执行器,负责执行一组Task的工作单元,类似Yarn中的TaskTracker(Yarn中的JobTracker和Yarn中的TaskTracker通过RPC进行通讯);二.

大家好,又见面了,我是你们的朋友全栈君。

先后看完了TaskGroupContainer 和 JobContainer,梳理下他们的关系与职责;


一、各自职责

  1. JobContainer: Job执行器,负责Job全局拆分、调度、前置语句和后置语句等工作的工作单元。类似Yarn中的JobTracker

  2. TaskGroupContainer: TaskGroup执行器,负责执行一组Task的工作单元,类似Yarn中的TaskTracker(Yarn中的JobTracker和Yarn中的TaskTracker通过RPC进行通讯);


二、相互关系

  1. TaskGroupContainer向JobContainer上报信息,称JobContainer是TaskGroupContainer的上级;
  2. Task向TaskGroupContainer上报信息,称TaskGroupContainer是Task的上级。

在这里插入图片描述


三、TaskGroupContainer向JobContianer汇报过程

汇报源码逻辑是在TaskGroupContainer#reportTaskGroupCommunication这个方法中,方法的两个形参分别为lastTaskGroupContainerCommunication为上次汇报的信息,每次做数据统计的时候需要将当前communication的数据和lastTaskGroupContainerCommunication进行合并;taskCount为该TaskGroup的所有的任务数。

  1. 收集当前TaskGroupContainer对应所有Task的的communication,然后将其合并成一个communication。
    具体合并代码为步骤1,主要逻辑是在Communication#mergeFrom,它的主要功能将两communication的变量合并。主要关注下两个communication的状态合并,可以看到只要该TaskGroup中有一个Task的状态是FAILED或者KILLED就会将整个TaskGroup的状态标记为FAILED,当且仅当所有的任务的状态是SUCCEEDED,该TaskGroup的状态才能标记为SUCCEEDED。

  2. 生成新的reportCommunication作为该TaskGroupContainer上报给JobContianer的communication,
    主要是生成一些技术统计,比方说当前已经导入的记录数和字节数等。

  3. 上报给JobContianer,主要代码见步骤2,将该TaskGroupContainer最新的communication更新到StandAloneJobContainerCommunicator 能够够的得到的地方,即全局变量LocalTGCommunicationManager#taskGroupCommunicationMap中。


  private Communication reportTaskGroupCommunication(Communication lastTGContainerComm,int taskCnt) { 
   
    Communication nowTGContainerComm = this.containerCommunicator.collect();
    nowTGContainerComm.setTimestamp(System.currentTimeMillis());
    Communication reportComm = CommunicationTool
        .getReportCommunication(nowTGContainerComm, lastTGContainerComm, taskCnt);
    this.containerCommunicator.report(reportComm);
    return reportComm;
  }

步骤1


//AbstractCollector#collectFromTask
public Communication collectFromTask() { 
   
    Communication communication = new Communication();
    communication.setState(State.SUCCEEDED);

    for (Communication taskCommunication :
            this.taskCommunicationMap.values()) { 
   
    communication.mergeFrom(taskCommunication);
    }
    return communication;
}
Communication#mergeStateFrom
public synchronized State mergeStateFrom(final Communication otherComm) { 
   
        State retState = this.getState();
        if (otherComm == null) { 
   
            return retState;
        }

        if (this.state == State.FAILED || otherComm.getState() == State.FAILED
                || this.state == State.KILLED || otherComm.getState() == State.KILLED) { 
   
            retState = State.FAILED;
        } else if (this.state.isRunning() || otherComm.state.isRunning()) { 
   
            retState = State.RUNNING;
        }

        this.setState(retState);
        return retState;
}

步骤2


// StandaloneTGContainerCommunicator#report
public void report(Communication communication) { 
   
    super.getReporter().reportTGCommunication(super.taskGroupId, communication);
}

// ProcessInnerReporter
public void reportTGCommunication(Integer taskGroupId, Communication communication) { 
   
    LocalTGCommunicationManager.updateTaskGroupCommunication(taskGroupId, communication);
}

// LocalTGCommunicationManager#updateTaskGroupCommunication
public static void updateTaskGroupCommunication(final int taskGroupId,
                                                    final Communication communication) { 
   
        Validate.isTrue(taskGroupCommunicationMap.containsKey(
                taskGroupId), String.format("taskGroupCommunicationMap中没有注册taskGroupId[%d]的Communication," +
                "无法更新该taskGroup的信息", taskGroupId));
        taskGroupCommunicationMap.put(taskGroupId, communication);
}


四、TaskGroupContainer向JobContianer汇报时机

TaskGroupContainer向JobContainer汇报该TaskGroupContainer的执行情况的时机均在TaskGroupContainer#start中。

1、当前TaskGroup中有状态为FAILED或者KILLED的Task

如果一个Task只能执行一次(默认是1次,没有做重试)且该Task被标记为FAILED或者KILLED,马上将failedOrKilled这个变量标记为true并执行汇报逻辑。这种情况下除了汇报之后,还会抛出一个运行时异常,结束执行当前TaskGroupContainer的线程(TaskGroupContianer是在线程池中执行的)。


if (failedOrKilled) { 
   
    lastTaskGroupContainerCommunication = reportTaskGroupCommunication(
            lastTaskGroupContainerCommunication, taskCountInThisTaskGroup);

    throw DataXException.asDataXException(
        FrameworkErrorCode.PLUGIN_RUNTIME_ERROR, lastTaskGroupContainerCommunication.getThrowable());
}

2、上次失败的Task仍未结束

如果一个Task标记为FAILED或者KILLED,但是有重试逻辑就不会执行上面第1步的逻辑,而是会调用当前的Task对应TaskExecutor#shutdown,关闭当前的TaskExecutor。在调用TaskExecutor#shutdown一段时间发发现给TaskExecutor还没有关闭,触发下面逻辑,进行汇报的同时抛出异常。

if(now - failedTime > taskMaxWaitInMsec){ 
   
    markCommunicationFailed(taskId);
    reportTaskGroupCommunication(lastTaskGroupContainerCommunication, taskCountInThisTaskGroup);
    throw DataXException.asDataXException(CommonErrorCode.WAIT_TIME_EXCEED, "task failover等待超时");
}

3、TaskGroupContainer任务列表为空,所有任务都是成功执行, 搜集状态为SUCCEEDED

这个没什么好说的,该TaskGroup中所有的任务执行成功,该Job执行成功。

4、如果当前时间已经超出汇报时间的interval,那么我们需要马上汇报

可以理解为心跳了

5、TaskGroupContainer所在的线程正常结束时汇报一次

这个真没什么好说的了


五、JobContainer收到汇报之后的处理

JobContainer的处理逻辑是在dataX所在JVM的主线程中,具体是在AbstractScheduler#schedule中。

  1. 每隔一段时间,合并所有TaskGoupContianer汇报的信息,具体合并的逻辑和TaskGoupContianer合并Task的汇报信息差不多;

  2. 正常结束就正常退出;

  3. 处理isJobKilling,StandAloneScheduler并没有提供kill接口,咱不管;

  4. 重点关注下FAILED的逻辑,直接关闭当前Scheduler的线程池并在主线程中抛出异常,整个dataX进程退出。

// AbstractScheduler#schedule
public void schedule(List<Configuration> configurations) { 

...
...
Communication lastJobContainerCommunication = new Communication();
long lastReportTimeStamp = System.currentTimeMillis();
try { 

while (true) { 

Communication nowJobContainerCommunication = this.containerCommunicator.collect();
nowJobContainerCommunication.setTimestamp(System.currentTimeMillis());
LOG.debug(nowJobContainerCommunication.toString());
//汇报周期
long now = System.currentTimeMillis();
if (now - lastReportTimeStamp > jobReportIntervalInMillSec) { 

Communication reportCommunication = CommunicationTool
.getReportCommunication(nowJobContainerCommunication, lastJobContainerCommunication, totalTasks);
this.containerCommunicator.report(reportCommunication);
lastReportTimeStamp = now;
lastJobContainerCommunication = nowJobContainerCommunication;
}
errorLimit.checkRecordLimit(nowJobContainerCommunication);
if (nowJobContainerCommunication.getState() == State.SUCCEEDED) { 

LOG.info("Scheduler accomplished all tasks.");
break;
}
if (isJobKilling(this.getJobId())) { 

dealKillingStat(this.containerCommunicator, totalTasks);
} else if (nowJobContainerCommunication.getState() == State.FAILED) { 

dealFailedStat(this.containerCommunicator, nowJobContainerCommunication.getThrowable());
}
Thread.sleep(jobSleepIntervalInMillSec);
}
} catch (InterruptedException e) { 

// 以 failed 状态退出
LOG.error("捕获到InterruptedException异常!", e);
throw DataXException.asDataXException(
FrameworkErrorCode.RUNTIME_ERROR, e);
}
}
// ProcessInnerScheduler#dealFailedStat
public void dealFailedStat(AbstractContainerCommunicator frameworkCollector, Throwable throwable) { 

this.taskGroupContainerExecutorService.shutdownNow();
throw DataXException.asDataXException(
FrameworkErrorCode.PLUGIN_RUNTIME_ERROR, throwable);
}

注:

  1. 对源码进行略微改动,主要修改为 1 阿里代码规约扫描出来的,2 clean code;

  2. 所有代码都已经上传到github(master分支和dev),可以免费白嫖

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/145686.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)
blank

相关推荐

  • SVN安装与使用_刚安装ETC不能马上使用吗

    SVN安装与使用_刚安装ETC不能马上使用吗SVN(Subversion)是近年来崛起的版本管理工具,在当前的开源项目里(J2EE),几乎95%以上的项目都用到了SVN。Subversion项目的初衷是为了替换当年开源社区最为流行的版本控制软件CVS,在CVS的功能的基础上有很多的提升同时也能较好的解决CVS系统的一些不足。svn是基于客户/服务器模式,所以分客户端和服务器端,作为开发人员,自己的电脑上只需要安装客户端,又…

  • 16天记住7000考研英语单词_最难记的考研单词

    16天记住7000考研英语单词_最难记的考研单词

    16天记住7000考研单词(第九天)
    401.Thebelovednovelistputherlovelyglovesabovethestove.
    敬爱的小说家把她美丽的手套放在火炉上方。
    402.It’sprovedthattheapproverimprovedwaterproofroof.
    经证实,赞同者改善了防水屋顶.
    403.Inthereaction,thefractionac

  • redis的部署方式,主从,集群(docker搭建mysql主从)

    Redis集群搭建——主从模式前言一、什么是主从模式?二、主从模式集群的搭建案例总结前言一般情况下,为了减轻数据库的访问压力,我们会把热点数据保存在内存中而不是直接从后端数据库中读取。Redis虽然是一个极其优秀的非关系型数据库,但是在大型网站应用,热点数据的并发访问量达到百万千万是很正常的,这个时候单个redis就不能够保证数据量的访问和存储。这个时候我们就可以搭建redis集群,可以保证数据的分散存储与数据的一致性,实现redis的高可用,发生故障时保证程序的正常运行与数据的保存。Redis有几

  • quotedStr() 应用–DELPHI「建议收藏」

    quotedStr() 应用–DELPHI「建议收藏」varcsql,cstr:string;begin//quotedStr(edit1.text)以说是在sql查询分析器里给字符串edit1.text加个单引号,可以完全抛开delphi的语法.cStr…

    2022年10月17日
  • ManualResetEvent用法「建议收藏」

    ManualResetEvent用法「建议收藏」ManualResetEvent用法

  • PyCharm如何安装torch

    PyCharm如何安装torch运行Pycharm中的代码时候提示ModuleNotFoundError:Nomodulenamed‘torch’。试了很多种方法都不行,然后进入官网查了下具体的安装方法,附上网址添加链接描述。摘取一段放在这里供大家参考。#CUDA10.0pipinstalltorch===1.2.0torchvision===0.4.0-fhttps://download.pytorch.org/whl/torch_stable.html#CUDA9.2pipinstalltor

    2022年10月29日

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号