概述
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
的类。以后我们会单独分析。