Flink架构分析之Standalone模式启动流程

概述

FLIP6 对Flink架构进行了改进,引入了Dispatcher组件集成了所有任务共享的一些组件:SubmittedJobGraphStore,LibraryCacheManager等,为了保证高可用,存在多个Dispatcher进行Master选举,同时Dispatcher必须把JobGraphs和提交job的相关jar包存储到持久化仓库中,保证failover后能恢复已经运行的任务。

Flink架构分析之Standalone模式启动流程

本文基于flink1.7.2进行分析,解析standalone模式的启动流程与架构,首先下载该版本的源代码发行包:flink-1.7.2-src 解压并编译

cd flink-1.7.2
mvn clean install -DskipTests
cd flink-dist
mvn clean install

Jobmanager

cd target/flink-1.7.2-bin/flink-1.7.2/bin
cat start-cluster.sh

Flink架构分析之Standalone模式启动流程

这个脚本非常清晰,先启动jobmanager然后启动taskmanager。在此我们只关注启动jobmanager部分,jobmanager的启动与HA配置有关,如果没有配置HA模式,那么仅仅是调用jobmanager.sh脚本在本机启动一个进程。

如果配置了HA模式,那么又会依据conf/master文件的配置,如果所有的host配置都是localhost或者127.0.0.1则在本机启动多个jobmanager进程,即伪分布式模式。否则就ssh到不同的host节点上启动jobmanager进程。启动jobmanager均是调用jobmanager.sh脚本,接下来我们分析一下此脚本。

cat jobmanager.sh

Flink架构分析之Standalone模式启动流程

此脚本会根据flink配置文件conf/flink-conf.yaml中的配置设置jvm启动参数。并且把ENTRYPOINT变量设置为standalonesession然后调用flink-daemon.sh脚本。

cat flink-daemon.sh

Flink架构分析之Standalone模式启动流程

这个脚本把调用CLass类变量设置为org.apache.flink.container.entrypoint.StandaloneJobClusterEntryPoint,然后执行以下命令在后台启动jobmanager

$JAVA_RUN $JVM_ARGS ${FLINK_ENV_JAVA_OPTS} "${log_setting[@]}" -classpath "`manglePathList "$FLINK_TM_CLASSPATH:$INTERNAL_HADOOP_CLASSPATHS"`" ${CLASS_TO_RUN} "${ARGS[@]}" > "$out" 200<&- 2>&1 < /dev/null &

所以,新架构中org.apache.flink.container.entrypoint.StandaloneJobClusterEntryPoint代替了原来的org.apache.flink.runtime.jobmanager.JobManager作为新的入口类,我们由此入手分析改进后的Standalone模式的架构:

Flink架构分析之Standalone模式启动流程

新架构中,对session clustersingle job cluster这两种运行模式进行了代码重构,层次结构更加清晰:

  • ClusterEntrypoint:作为所有集群入口类的抽象父类,实现了通用的业务逻辑,并提供了createSerializableExecutionGraphStorecreateDispatcherResourceManagerComponentFactory两个抽象方法供子类实现。

  • SessionClusterEntrypoint:作为session cluster集群的直接父类,实现了ClusterEntrypointcreateSerializableExecutionGraphStore方法,采用FileArchivedExecutionGraphStore存储执行完成的任务。

  • JobClusterEntrypoint:作为single job cluster集群的直接父类,实现了ClusterEntrypointcreateSerializableExecutionGraphStore方法,采用MemoryArchivedExecutionGraphStore存储执行完成的任务。

  • standalone,yarn,mesos 分别继承SessionClusterEntrypointJobClusterEntrypoint并重写了createDispatcherResourceManagerComponentFactory方法实现了session clustersingle job cluster集群入口类。

DispatcherResourceManagerComponent

为了使结果更加清晰,更易于管理。新架构中用DispatcherResourceManagerComponent封装了Dispatcher,ResourceManager,WebMonitorEndpoint三个重要的对象.以后我们单独详细分析这三个类的功能。

Taskmanager

start-cluster.sh脚本的最后一行是调用TMSlaves start启动Taskmanager,TMSlaves是定义在bin/config.sh中的函数:

Flink架构分析之Standalone模式启动流程

调用taskmanager.sh脚本:

ENTRYPOINT=taskexecutor
"${FLINK_BIN_DIR}"/flink-daemon.sh $STARTSTOP $ENTRYPOINT "${ARGS[@]}"

最终调用了bin/flink-daemon.sh,并把启动入口类设置为org.apache.flink.runtime.taskexecutor.TaskManagerRunner:

Flink架构分析之Standalone模式启动流程

TaskManagerRunner包装了TaskExecutor类,TaskExecutor是真正管理和执行Task的类。以后我们会单独分析。

上一篇:为什么不能在scrollview中直接添加一个image,然后使animation.begin()??


下一篇:二十三、Django3.0学习之在apps下命令新建app应用