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

概述

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

本文基于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

这个脚本非常清晰，先启动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配置文件conf/flink-conf.yaml中的配置设置jvm启动参数。并且把ENTRYPOINT变量设置为standalonesession然后调用flink-daemon.sh脚本。

cat flink-daemon.sh

这个脚本把调用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模式的架构:



新架构中，对session cluster和single job cluster这两种运行模式进行了代码重构，层次结构更加清晰：

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

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

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

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

DispatcherResourceManagerComponent

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

Taskmanager

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

调用taskmanager.sh脚本:

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

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



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