Flink Task退出流程与Failover机制

这篇具有很好参考价值的文章主要介绍了Flink Task退出流程与Failover机制。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

1 TaskExecutor端Task退出逻辑

Task.doRun() 引导Task初始化并执行其相关代码的核心方法,
构造并实例化Task的可执行对象: AbstractInvokable invokable。
调用 AbstractInvokable.invoke() 开始启动Task包含的计算逻辑。
Flink Task退出流程与Failover机制,flink技术原理,flink,大数据
Flink Task退出流程与Failover机制,flink技术原理,flink,大数据

当AbstractInvokable.invoke()执行退出后,根据退出类型执行相应操作:

  1. 正常执行完毕退出:输出ResultPartition缓冲区数据,并关闭缓冲区,标记Task为Finished;
  2. 取消操作导致退出:标记Task为CANCELED,关闭用户代码;
  3. AbstractInvokable.invoke()执行过程中抛出异常退出:标记Task为FAILED,关闭用户代码,记录异常;
  4. AbstractInvokable.invoke()执行过程中JVM抛出错误:强制终止虚拟机,退出当前进程。

紧接着释放Task相关的网络、内存、文件系统资源。最后通过Task->TaskManager->JobMaster的传递链路将Task的终止状态通知给Leader JobMaster线程。

Task.notifyFinalState() -> TaskManagerActions.updateTaskExecutionState(TaskExecutionState) -> JobMasterGateway.updateTaskExecutionState(TaskExecutionState)

  • TaskExecutionState携带的关键信息:
TaskExecutionState {
    JobID // 任务ID
    ExecutionAttemptID  // Task执行的唯一ID,标示每次执行
    ExecutionState  // 枚举值,Task执行状态
    SerializedThrowable // 若Task抛出异常,该字段记录异常堆栈信息
    ...
}
  • Task 执行状态转换:
   CREATED  -> SCHEDULED -> DEPLOYING -> RUNNING -> FINISHED
      |            |            |          |
      |            |            |   +------+
      |            |            V   V
      |            |         CANCELLING -----+----> CANCELED
      |            |                         |
      |            +-------------------------+
      |
      |                                   ... -> FAILED
      V
  RECONCILING  -> RUNNING | FINISHED | CANCELED | FAILED

2 JobMaster端failover流程

2.1 Task Execute State Handle

JobMaster收到TaskManager通过rpc发送的task执行状态变更信息,将通知当前Flink作业的调度器(SchedulerNG)处理,因为都是通过同个线程调用,后续对ExecutionGraph(运行时执行计划)、failover计数等有状态实例的read/write操作都不会出现线程安全问题。

JobMaster的核心处理逻辑在SchedulerBase.updateTaskExecutionState(TaskExecutionStateTransition) 中(TaskExecutionStateTransition主要是TaskExecutionState的可读性封装)。
处理逻辑:尝试将收到的Task执行状态信息更新到ExecutionGraph中。若更新成功且target状态为FINISHED,根据具体的SchedulingStrategy实现策略,选择可消费的结果分区并调度相应的消费者Task;若更新成功且target状态为FAILED,进入具体的failover流程。

  • SchedulerBase.updateTaskExecutionState(TaskExecutionStateTransition) :
    public final boolean updateTaskExecutionState(
            final TaskExecutionStateTransition taskExecutionState) {
        final Optional<ExecutionVertexID> executionVertexId =
                getExecutionVertexId(taskExecutionState.getID());

        boolean updateSuccess = executionGraph.updateState(taskExecutionState);

        if (updateSuccess) {
            checkState(executionVertexId.isPresent());
            if (isNotifiable(executionVertexId.get(), taskExecutionState)) {
                updateTaskExecutionStateInternal(executionVertexId.get(), taskExecutionState);
            }
            return true;
        } else {
            return false;
        }
    }

  • ExecutionGraph.updateState(TaskExecutionStateTransition): 在当前的物理执行拓扑中找不到目标ExecutionAttemptID 时,将更新失败。需要注意的是这个ID用于唯一标示一个Execution,而Execution则代表ExecutionVertex(代表拓扑顶点的一个subTask计划)的一次执行实例,ExecutionVertex可以重复多次执行。这意味着当有subTask重新运行,currentExecutions将不再持有上一次执行的ID信息。
   /**
     * Updates the state of one of the ExecutionVertex's Execution attempts. If the new status if
     * "FINISHED", this also updates the accumulators.
     *
     * @param state The state update.
     * @return True, if the task update was properly applied, false, if the execution attempt was
     *     not found.
     */
    public boolean updateState(TaskExecutionStateTransition state) {
       assertRunningInJobMasterMainThread();
        final Execution attempt = currentExecutions.get(state.getID());
        if (attempt != null) {
            try {
                final boolean stateUpdated = updateStateInternal(state, attempt);
                maybeReleasePartitions(attempt);
                return stateUpdated;
            } catch (Throwable t) {
                ......
                return false;
            }
        } else {
            return false;
        }
    }

  • JobMaster: 负责一个任务拓扑的中心操作类,涉及作业调度,资源管理,对外通讯等…

  • SchedulerNG:负责调度作业拓扑。所有对该类对象方法的调用都会通过ComponentMainThreadExecutor触发,将不会出现并发调用的情况。

  • ExecutionGraph: 当前执行拓扑的中心数据结构,协调分布在各个节点上的Execution。描述了整个任务的各个SubTask及其分区数据,并与其保持通讯。

2.2 Job Failover

2.2.1 Task Failure Handle

  • Task异常的主要流程在 DefaultScheduler.handleTaskFailure(ExecutionVertexID, Throwable), 根据RestartBackoffTimeStrategy判断是重启还是failed-job;根据FailoverStrategy选择需要重启的SubTask;最后根据任务当前的SchedulingStrategy执行相应的调度策略重启相应的Subtask。
    private void handleTaskFailure(
            final ExecutionVertexID executionVertexId, @Nullable final Throwable error) {
        // 更新当前任务异常信息
        setGlobalFailureCause(error);
        // 如果相关的算子(source、sink)存在coordinator,同知其进一步操作
        notifyCoordinatorsAboutTaskFailure(executionVertexId, error);
        // 应用当前的restart-stratege并获取FailureHandlingResult
        final FailureHandlingResult failureHandlingResult =
                executionFailureHandler.getFailureHandlingResult(executionVertexId, error);
        // 根据结果重启Task或将任务失败
        maybeRestartTasks(failureHandlingResult);
    }


    public class FailureHandlingResult {
        //恢复所需要重启的所有SubTask
          Set<ExecutionVertexID> verticesToRestart;
        //重启延迟
          long restartDelayMS;
        //万恶之源
          Throwable error;
        //是否全局失败
          boolean globalFailure;
    }

  • ExecutionFailureHandler:处理异常信息,根据当前应用策略返回异常处理结果。
    public FailureHandlingResult getFailureHandlingResult(
            ExecutionVertexID failedTask, Throwable cause) {
        return handleFailure(
                cause, 
                failoverStrategy.getTasksNeedingRestart(failedTask, cause),  // 选择出需要重启的SubTask
                false); 
    }

    private FailureHandlingResult handleFailure(
            final Throwable cause,
            final Set<ExecutionVertexID> verticesToRestart,
            final boolean globalFailure) {

        if (isUnrecoverableError(cause)) {
            return FailureHandlingResult.unrecoverable(
                    new JobException("The failure is not recoverable", cause), globalFailure);
        }

        restartBackoffTimeStrategy.notifyFailure(cause);
        if (restartBackoffTimeStrategy.canRestart()) {
            numberOfRestarts++;

            return FailureHandlingResult.restartable(
                    verticesToRestart, restartBackoffTimeStrategy.getBackoffTime(), globalFailure);
        } else {
            return FailureHandlingResult.unrecoverable(
                    new JobException(
                            "Recovery is suppressed by " + restartBackoffTimeStrategy, cause),
                    globalFailure);
        }
    }

  • FailoverStrategy: 故障转移策略。

    • RestartAllFailoverStrategy: 使用该策略,当出现故障,将重启整个作业,即重启所有Subtask。
    • RestartPipelinedRegionFailoverStrategy:当出现故障,重启故障出现Subtask所在的的Region。
    • RestartBackoffTimeStrategy: 当Task发生故障时,决定是否重启以及重启的延迟时间。
  • FixedDelayRestartBackoffTimeStrategy:允许任务以指定延迟重启固定次数。

    • FailureRateRestartBackoffTimeStrategy:允许在固定失败频率内,以固定延迟重启。
    • NoRestartBackoffTimeStrategy:不重启。
  • SchedulingStrategy: Task执行实例的调度策略

    • EagerSchedulingStrategy: 饥饿调度,同时调度所有Task。
    • LazyFromSourcesSchedulingStrategy:当消费的分区数据都准备好后才开始调度其后续Task,用于批处理任务。
    • PipelinedRegionSchedulingStrategy:以pipline链接的Task为一个Region,作为其调度粒度。

2.2.2 Restart Task


    private void maybeRestartTasks(final FailureHandlingResult failureHandlingResult) {
        if (failureHandlingResult.canRestart()) {
            restartTasksWithDelay(failureHandlingResult);
        } else {
            failJob(failureHandlingResult.getError());
        }
    }

    private void restartTasksWithDelay(final FailureHandlingResult failureHandlingResult) {
        final Set<ExecutionVertexID> verticesToRestart =
                failureHandlingResult.getVerticesToRestart();

        final Set<ExecutionVertexVersion> executionVertexVersions =
                new HashSet<>(
                        executionVertexVersioner
                                .recordVertexModifications(verticesToRestart)
                                .values());
        final boolean globalRecovery = failureHandlingResult.isGlobalFailure();

        addVerticesToRestartPending(verticesToRestart);

        // 取消所有需要重启的SubTask
        final CompletableFuture<?> cancelFuture = cancelTasksAsync(verticesToRestart);

        delayExecutor.schedule(
                () ->
                        FutureUtils.assertNoException(
                                cancelFuture.thenRunAsync(   // 停止后才能重新启动
                                        restartTasks(executionVertexVersions, globalRecovery), 
                                        getMainThreadExecutor())),
                failureHandlingResult.getRestartDelayMS(),
                TimeUnit.MILLISECONDS);
    }


2.2.3 Cancel Task:

  • 取消正在等待Slot分配的SubTask,若已经处于部署/运行状态,则需要通知TaskExecutor执行停止操作并等待操作完成。
    private CompletableFuture<?> cancelTasksAsync(final Set<ExecutionVertexID> verticesToRestart) {
        // clean up all the related pending requests to avoid that immediately returned slot
        // is used to fulfill the pending requests of these tasks
        verticesToRestart.stream().forEach(executionSlotAllocator::cancel); // 取消可能正处于等待分配Slot的SubTask

        final List<CompletableFuture<?>> cancelFutures =
                verticesToRestart.stream()
                        .map(this::cancelExecutionVertex) // 开始停止SubTask
                        .collect(Collectors.toList());

        return FutureUtils.combineAll(cancelFutures);
    }

    public void cancel() {
        while (true) { // 状态变更失败则重试
            ExecutionState current = this.state;
            if (current == CANCELING || current == CANCELED) {
                // already taken care of, no need to cancel again
                return;
            }
            else if (current == RUNNING || current == DEPLOYING) {
                // 当前状态设为CANCELING,并向TaskExecutor发送RPC请求停止SubTask
                if (startCancelling(NUM_CANCEL_CALL_TRIES)) {
                    return;
                }
            } else if (current == FINISHED) {
                // 即使完成运行,后续也无法消费,释放分区结果
                sendReleaseIntermediateResultPartitionsRpcCall();
                return;
            } else if (current == FAILED) {
                return;
            } else if (current == CREATED || current == SCHEDULED) {
                // from here, we can directly switch to cancelled, because no task has been deployed
                if (cancelAtomically()) {
                    return;
                }
            } else {
                throw new IllegalStateException(current.name());
            }
        }
    }

  • 操作完毕后又会执行Task退出流程通知ExecutionGraph执行相应数据更新: ExecutionGraph.updateState(TaskExecutionStateTransition)->ExecutionGraph.updateStateInternal(TaskExecutionStateTransition, Execution) -> Execution.completeCancelling(…) -> Execution.finishCancellation(boolean) -> ExecutionGraph.deregisterExecution(Execution) 。在deregisterExecution操作会在currentExecutions中移除已停止的执行ExecutionTask。

2.2.4 Start Task

        private Runnable restartTasks(
                final Set<ExecutionVertexVersion> executionVertexVersions,
                final boolean isGlobalRecovery) {
            return () -> {
                final Set<ExecutionVertexID> verticesToRestart =
                        executionVertexVersioner.getUnmodifiedExecutionVertices(
                                executionVertexVersions);
    
                removeVerticesFromRestartPending(verticesToRestart);
                // 实例化新的SubTask执行实例(Execution)
                resetForNewExecutions(verticesToRestart);
    
                try {
                    // 恢复状态
                    restoreState(verticesToRestart, isGlobalRecovery);
                } catch (Throwable t) {
                    handleGlobalFailure(t);
                    return;
                }
                // 开始调度,申请Slot并部署
                schedulingStrategy.restartTasks(verticesToRestart);
            };
        }

3 Task失败的自动重启策略

Task 级别的故障重启,是系统自动进行的;

Flink Task退出流程与Failover机制,flink技术原理,flink,大数据
Flink Task退出流程与Failover机制,flink技术原理,flink,大数据
Flink Task退出流程与Failover机制,flink技术原理,flink,大数据
Flink Task退出流程与Failover机制,flink技术原理,flink,大数据
Flink Task退出流程与Failover机制,flink技术原理,flink,大数据文章来源地址https://www.toymoban.com/news/detail-836434.html

到了这里,关于Flink Task退出流程与Failover机制的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Flink源码-Task执行

    上一节我们分析到了Execution的生成,然后调用taskManagerGateway.submitTask方法提交task,提交的时候会将executionVertex封装成TaskDeploymentDescriptor,task的提交与执行涉及到了flink多个组件的配合,之前没有详细讲过,可能有的小伙伴有点不太清楚,这里我们花点时间介绍一下。 1.JobManager

    2024年02月03日
    浏览(37)
  • Flink之Task重启策略

    Task重启策略 1 策略API noRestart 无参数,task失败后不重启,整个job同时失败,默认策略. 代码示例 fixedDelayRestart 参数 注释 restartAttempts 最大重启次数 delayBetweenAttempts 重启时间间隔 代码示例 exponentialDelayRestart 参数 注释 initialBackoff 重启间隔惩罚时长初始值(重启延迟时间) maxBackoff 重

    2024年02月03日
    浏览(38)
  • 深入理解 Flink(八)Flink Task 部署初始化和启动详解

    核心入口: 部署 Task 链条:JobMaster -- DefaultScheduler -- SchedulingStrategy -- ExecutionVertex -- Execution -- RPC请求 -- TaskExecutor JobMaster 向 TaskExecutor 发送 submitTask() 的 RPC 请求,用来部署 StreamTask 运行。TaskExecutor 接收到 JobMaster 的部署 Task 运行的 RPC 请求的时候,就封装了一个 Task 抽象,然

    2024年01月17日
    浏览(75)
  • Flink checkpoint操作流程详解与报错调试方法汇总,增量checkpoint原理及版本更新变化,作业恢复和扩缩容原理与优化

    本文主要参考官方社区给出的checkpoint出错类型和种类,以及查找报错的方法。 主要分为两种 Checkpoint Decline 与 Checkpint Expire 两种类型 下面分开讨论 从业务上来讲,Checkpoint 失败可能有较多的影响。 Flink 恢复时间长,会导致服务可用率降低。 非幂等或非事务场景,导致大量业

    2024年02月22日
    浏览(45)
  • 【flink】Task 故障恢复详解以及各重启策略适用场景说明

    当 Task 发生故障时,Flink 可以重启出错的 Task 以及其他受到影响的 Task ,以使得作业恢复到正常执行状态。 Flink 通过重启策略和故障恢复策略来控制 Task 重启: 重启策略决定是否可以重启以及重启的间隔; 故障恢复策略决定 哪些 Task 需要重启 。   参数 restart-strategy 定义了

    2024年02月04日
    浏览(40)
  • (增加细粒度资源管理)深入理解flink的task slot相关概念

    之前对flink的task slot的理解太浅了,重新捋一下相关知识点 我们知道,flink中每个TaskManager都是一个 JVM 进程,可以在单独的线程中执行一个或多个 subtask(线程)。 但是TaskManager 的计算资源是有限的,并不是所有任务都可以放在同一个 TaskManager 上并行执行。并行的任务越多

    2024年03月11日
    浏览(41)
  • 【Redis深度专题】「核心技术提升」探究Redis服务启动的过程机制的技术原理和流程分析的指南(集群指令分析—上篇)

    Redis Cluster提供了一套完整的功能技术,使得Redis能够以分布式的方式运行,并具备高可用性、容错性和扩展性。通过自动发现、主从选举、在线分片等机制,Redis Cluster能够自动管理集群中的节点,并保证数据的一致性和可靠性。同时,基于配置文件和转向机制,Redis Cluster能

    2024年02月14日
    浏览(52)
  • Flink 启动就报错,但exception没提示。其中一个task failure 该怎么办?

    最近我在生产又遇到一个问题,就是消费着一段时间之后,忽然就不再消费了,但也不报错。观察了几次,我发现时间基本是停留在上下班高峰期数据量最大的时候。我主观猜测可能是同时间进来的数据过多,处理不来导致的。但这个问题我还没来的及思考怎么处理,因此我

    2024年02月16日
    浏览(55)
  • Flink 学习八 Flink 容错机制 & checkpoint & savepoint

    https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/concepts/stateful-stream-processing/ 上一节讲述 状态后端 ;Flink是一个 带状态stateful 的数据处理系统,在处理数据的过程中,各个算子的记录的状态会随着算子处理的状态而改变 ; 状态后端 负责将状态保存在内存或外部持久化存储中 (内存

    2024年02月09日
    浏览(77)
  • 【Flink】Flink提交流程

    我们通常在学习的时候需要掌握大数据组件的原理以便更好的掌握这个大数据组件,Flink实际生产开发过程中最常见的就是提交到yarn上进行调度,模式使用的 Per-Job模式,下面我们就给大家讲下Flink提交Per-Job任务到yarn上的流程,流程图如下  (1)客户端将作业提交给 YARN 的资

    2024年02月11日
    浏览(32)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包