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日
    浏览(39)
  • Flink之Task重启策略

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

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

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

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

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

    2024年02月22日
    浏览(47)
  • Flink 中kafka broker缩容导致Task一直重启

    Flink版本 1.12.2 Kafka 客户端 2.4.1 在公司的Flink平台运行了一个读Kafka计算DAU的流程序,由于公司Kafka的缩容,直接导致了该程序一直在重启,重启了一个小时都还没恢复(具体的所容操作是下掉了四台kafka broker,而当时flink配置了12台kafka broker),当时具体的现场如下: 当时Fl

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

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

    2024年02月04日
    浏览(42)
  • (增加细粒度资源管理)深入理解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日
    浏览(55)
  • Flink 启动就报错,但exception没提示。其中一个task failure 该怎么办?

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

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

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

    2024年02月09日
    浏览(79)

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

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

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

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

二维码1

领取红包

二维码2

领红包