Apache Flink源码解析 （三）Flink On Yar

杂谈

有一段时间没有写技术博客了，正好之前花了一周的时间解决了一个Flink在Yarn上部署的问题，也将Flink在Yarn上运行的机制与源码了解了个大概。Flink在Yarn上部署涉及到的东西比较多，肯定不会一次写完，我应该会分几篇文章按几个模块来详细阐述，这样看起来也会比较轻松一些，而版本是基于Flink刚发布的release-1.5.0。

概述

• Flink在Yarn上有两种模式，一种是cluster模式，即像Yarn申请一定量的资源，有点类似于Standalone模式，当然我觉得缺点应该很明显（这里我也不是很肯定，因为我使用的不是这种模式），就是资源的浪费，以及扩容的时候需要重启，影响业务。另一种是Single Job模式，即将一个单独的Job提交到Yarn集群，由Yarn来根据配置分配Container。在使用这个模式的时候踩了一些坑，以后有机会细讲。

• 在这篇文章中我假定大家对Yarn已经有了一定的认识，不会在Yarn的概念上进行展开描述。

ApplicationMaster

• 首先是JobManager和ApplicationMaster。在Flink on Yarn中，JobManager和ApplicationMaster是在同一个jvm进程中的，这个进程的入口就是YarnSessionClusterEntrypoint类。首先来看一下这个类，它继承了SessionClusterEntrypoint类。

  
  YarnSessionClusterEntrypointUML
  

  SessionClusterEntrypoint是一个抽象类，它又继承了ClusterEntrypoint这个抽象类。

• 那么这三个类到底是干什么的呢？首先是ClusterEntrypoint，它封装了Cluster启停的逻辑，还有根据配置文件来创建RpcService，HaService, HeartbeatService, MetricRegistry等等服务的逻辑，同时它也提供了几个抽象方法给不同的模式下的特定的ClusterEntrypoint来实现。这四个方法分别是createDispatcher，createResourceManager， createRestEndpoint，createSerializableExecutionGraphStore。关于Dispatcher，ResourceManager和SerializableExecutionGraphStore，后面会有单独的文章来详细讲述，现在可以顾名思义，根据它的名字大致就能知道它们提供了怎样的功能。

• 而SessionClusterEntrypoint继承了这个类，并且实现了部分方法createSerializableExecutionGraphStore，createRestEndpoint，createDispatcher。为什么没有实现createResourceManager方法呢？从SessionClusterEntrypoint的子类就可以看出来，分别是YarnSessionClusterEntrypoint，MesosSessionClusterEntrypoint，StandaloneSessionClusterEntrypoint，这三者都是跟如何去调度资源相关的，所以createResourceManager由这些子类来实现。

• 这里我只会分析YarnSessionClusterEntrypoint。在这个类中，除了提供启动的main函数以外，最重要的是实现createResourceManager方法，在这个方法中，直接实例化了一个YarnResourceManager的对象并返回。

  YarnSessionClusterEntrypoint启动流程

  • 下面来看一下YarnSessionClusterEntrypoint的启动流程。首先入口是main函数，在main函数中新建了YarnSessionClusterEntrypoint的一个对象。

    
    main method of YarnSessionClusterEntrypoint

  • 构造这个对象的构造参数是配置信息和目录。紧接着就调用了startCluster方法。之前也提到，Cluster的启停逻辑是在父类ClusterEntrypoint中。至此，YarnSessionClusterEntrypoint的最重要的任务已经完成。

    
    startCluster method of ClusterEntrypoint
    

    在startCluster方法中，异步的调用了runCluster方法。

runCluster method of ClusterEntrypoint

• 在runCluster函数中，主要做了三件事：

  • 第一件事是根据配置文件初始化RpcService，HaService, HeartbeatService, MetricRegistry，BlobServer，ResourceManager，SerializableExecutionGraphStore等模块，并将它们赋值给ClusterEntrypoint中相应的instance variables。其中createSerializableExecutionGraphStore，createRestEndpoint，createDispatcher的逻辑由SessionClusterEntrypoint实现，createResourceManager的逻辑由YarnSessionClusterEntrypoint实现。

  • 第二件事就是在startClusterComponents函数中启动这些服务.其中RpcService负责各个模块之间的rpc调用，本质上是基于Akka的（关于Akka我在之前的文章中概述过）。ResourceManager主要是负责与Yarn的ResourceManager进行交互，通过Yarn提供的AMRMAsyncClient，进行一些资源分配与释放的操作。HaService的任务主要有ResourceManager，JobManager，Dispatcher，WebMonitor的Leader选举，checkpoint的元数据的持久化以及checkpoint的注册与跟踪，Blob数据的持久化，任务的状态监控，总而言之就是服务高可用相关的功能。HeartbeatService是负责心跳的发送与接收，被多个模块用来监控其他节点是否丢失。MetricRegistry负责指标的监控。BlobServer负责处理对Blob数据的请求与返回。而SerializableExecutionGraphStore则负责储存序列化之后的执行计划。

  • 最后一件事就是就是启动Dispatcher, Dispatcher主要负责就收Client提交的任务，启动并将任务传递给JobManager，以及Master节点挂掉的失败恢复。
    *对于上述提到的各个模块，以后会有文章单独的去剖析。

总结

综上，YarnSessionClusterEntrypoint可以理解为Flink在Yarn上的ApplicationMaster，同时也是JobManager。它们之间分属两个线程，之间的交互通过Akka的消息驱动的模式来实现任务调度与资源分配的分离，而对应的JobManager与ResourceManager也有相应的子模块组成。