中英文网站开发公司,签订网站建设合同,天行健君子以自强不息网站建设,网站开发 高级认证前言
JobManager是Flink的核心进程#xff0c;主要负责Flink集群的启动和初始化#xff0c;包含多个重要的组件(JboMaster#xff0c;Dispatcher#xff0c;WebEndpoint等)#xff0c;本篇文章会基于源码分析JobManagr的启动流程#xff0c;对其各个组件进行介绍#x…前言
JobManager是Flink的核心进程主要负责Flink集群的启动和初始化包含多个重要的组件(JboMasterDispatcherWebEndpoint等)本篇文章会基于源码分析JobManagr的启动流程对其各个组件进行介绍希望对JobManager有一个更全面的了解。
集群启动模式
ClusterEntryPoint是Flink集群的入口点的基类该类是抽象类类继承关系UML图如下 通过上图可知道Flink有3种集群模式
Flink Session集群
根据不同的资源管理器有3个不同的子类
StandaloneSessionClusterEntrypoint Standalone session模式下集群的入口类KubernetesSessionClusterEntrypoint K8s session模式下集群的入口类YarnSessionClusterEntrypoint Yarn session模式下集群的入口类
集群生命周期
在 Flink Session 集群中客户端连接到一个预先存在的、长期运行的集群该集群可以接受多个作业提交。即使所有作业完成后集群和 JobManager仍将继续运行直到手动停止 session 为止。因此Flink Session 集群的寿命不受任何 Flink 作业寿命的约束。
资源隔离
Flink作业共享集群的ResourceManager和Dispacher等组件TaskManager slot 由 ResourceManager 在提交作业时分配并在作业完成时释放。由于所有作业都共享同一集群因此在集群资源方面存在一些竞争 — 例如提交工作阶段的网络带宽。此共享设置的局限性在于如果 TaskManager 崩溃则在此 TaskManager 上运行 task 的所有作业都将失败类似的如果 JobManager 上发生一些致命错误它将影响集群中正在运行的所有作业。
适用场景
因为组件共享session集群资源使用率高集群预先存在不需要额外申请资源适合一些比较小的不是长期运行的作业例如SQL预览交互式查询实时任务测试环境等
Flink Per Job集群
只要Yarn提供了继承的子类YarnJobClusterEntrypoint
集群生命周期
在 Flink Job 集群中可用的集群管理器例如 YARN用于为每个提交的作业启动一个集群并且该集群仅可用于该作业。一旦作业完成Flink Job 集群将被拆除。
资源隔离
每一个提交的Flink应用程序单独创建一套完整集群环境该Job独享使用的计算资源和组件服务。
使用场景
实时由于Per Job模式下用户应用程序的main方法在客户端执行生成JobGraph任务量大情况下存在性能瓶颈目前已被标记为废弃状态。
Flink Application集群
根据不同的资源管理器有3个不同的子类
StandaloneApplicationClusterEntryPoint Standalone Application模式下集群的入口类KubernetesApplicationClusterEntrypoint K8s Application模式下集群的入口类YarnApplicationClusterEntryPoint Yarn Application模式下集群的入口类
集群生命周期
Flink Application 集群是专用的 Flink 集群仅从 Flink 应用程序执行作业并且main方法在集群上而不是客户端上运行。应用程序逻辑和依赖打包成一个可执行的作业 JAR 中并且集群入口ApplicationClusterEntryPoint负责调用main方法来提取 JobGraph
资源隔离
每一个提交的Flink应用程序单独创建一套完整集群环境该Job独享使用的计算资源和组件服务。
使用场景
Application模式资源隔离性好Per Job模式的替换方案适合长期运行、具有高稳定性的大型作业
JobManager启动流程
JobManger启动流程在不同模式下基本相同Standalone模式可以在本地运行可以参考方便Debug因为使用Standalone模式的入口类StandaloneSessionClusterEntrypoint进行启动流程的分析。
main方法入口
public static void main(String[] args) {// 打印系统相关信息EnvironmentInformation.logEnvironmentInfo(LOG, StandaloneSessionClusterEntrypoint.class.getSimpleName(), args);//信号注册器注册系统级别的信号接收到系统级别终止信号优雅的关闭SignalHandler.register(LOG);//注册一个安全的钩子这样jvm停止之前会睡眠5s去释放资源5s之后强制关闭JvmShutdownSafeguard.installAsShutdownHook(LOG);// 解析命令行参数获取配置信final EntrypointClusterConfiguration entrypointClusterConfiguration ClusterEntrypointUtils.parseParametersOrExit(args,new EntrypointClusterConfigurationParserFactory(),StandaloneSessionClusterEntrypoint.class);//加载config.yaml构建Configuration对象Configuration configuration loadConfiguration(entrypointClusterConfiguration);StandaloneSessionClusterEntrypoint entrypoint new StandaloneSessionClusterEntrypoint(configuration);ClusterEntrypoint.runClusterEntrypoint(entrypoint);
}
主要步骤
打印系统信息。注册信号处理器注册系统级别的信号确保优雅关闭。注册一个安全的钩子这样jvm停止之前会睡眠5s去释放资源5s之后强制关闭。解析命令行参数加载配置文件。初始化 StandaloneSessionClusterEntrypoint。调用 ClusterEntrypoint#runClusterEntrypoint 方法启动集群。
ClusterEntrypoint#runClusterEntrypoint
public static void runClusterEntrypoint(ClusterEntrypoint clusterEntrypoint) {final String clusterEntrypointName clusterEntrypoint.getClass().getSimpleName();
try {//clusterEntrypoint.startCluster();
} catch (ClusterEntrypointException e) {LOG.error(String.format(Could not start cluster entrypoint %s., clusterEntrypointName),e);System.exit(STARTUP_FAILURE_RETURN_CODE);
}//无关代码 无需关注
}
核心步骤
调用 clusterEntrypoint.startCluster() 启动集群。
ClusterEntrypoint#startCluster
public void startCluster() throws ClusterEntrypointException {//无关代码 无需关注try {FlinkSecurityManager.setFromConfiguration(configuration);//插件管理类用来加载插件。插件加载两种方式。//1).通过如下参数配置FLINK_PLUGINS_DIR。//2).将插件jar包放入到plugins下PluginManager pluginManager PluginUtils.createPluginManagerFromRootFolder(configuration);//初始化文件系统的配置configureFileSystems(configuration, pluginManager);//初始化安全上下文环境 默认HadoopSecurityContextHadoop安全上下文//使用先前初始化的UGI(UserGroupInformation)和适当的安全凭据。比如Kerberos。//总结初始化安全环境创建安全环境的时候会做一系列的检查。SecurityContext securityContext installSecurityContext(configuration);ClusterEntrypointUtils.configureUncaughtExceptionHandler(configuration);//安全的情况下调用runCluster开始初始化组件securityContext.runSecured((CallableVoid)() - {runCluster(configuration, pluginManager);return null;});} catch (Throwable t) {//异常处理代码 无需关注}}
startCluster方法主要做了一些环境和配置初始化的工作
主要步骤
初始化插件管理器用来加载插件。初始化文件系统设置 例如 hdfs、本地file。此时只是初始化的配置。初始化安全环境。安全环境下调用 runCluster 方法。
ClusterEntrypoint#runCluster
private void runCluster(Configuration configuration, PluginManager pluginManager)throws Exception {synchronized (lock) {//初始化集群所需要的服务例如通信服务监控服务高可用服务等initializeServices(configuration, pluginManager);// write host information into configurationconfiguration.set(JobManagerOptions.ADDRESS, commonRpcService.getAddress());configuration.set(JobManagerOptions.PORT, commonRpcService.getPort());//创建Dispatcher和ResourceManger组件的工厂类final DispatcherResourceManagerComponentFactorydispatcherResourceManagerComponentFactory createDispatcherResourceManagerComponentFactory(configuration);//创建Dispatcher和ResourceManger组件clusterComponent dispatcherResourceManagerComponentFactory.create(configuration,resourceId.unwrap(),ioExecutor,commonRpcService,haServices,blobServer,heartbeatServices,delegationTokenManager,metricRegistry,executionGraphInfoStore,new RpcMetricQueryServiceRetriever(metricRegistry.getMetricQueryServiceRpcService()),failureEnrichers,this);//组件停止运行后的异步方法clusterComponent.getShutDownFuture().whenComplete(//代码省略)}}
主要步骤
1.初始化集群所需要的服务例如通信服务监控服务高可用服务等
2.创建Dispatcher和ResourceManger组件的工厂类
3.创建Dispatcher和ResourceManger组件
4.定义组件停止运行后的异步方法
总结
本篇文章分享了Flink任务的集群模式通过源码的方式分析了JobManger的启动流程后续会对JobManger相关的服务和组件进行更详细的分析。
