本篇内容主要讲解“Flink怎么执行用户程序”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“Flink怎么执行用户程序”吧!
创新互联主营恩施土家网站建设的网络公司,主营网站建设方案,手机APP定制开发,恩施土家h5成都微信小程序搭建,恩施土家网站营销推广欢迎恩施土家等地区企业咨询
以flink on yarn为例:
(1)在CliFrontend的main()方法中,生成GenericCLI、FlinkYarnSessionCli、DefaultCLI三种命令行对象,依次放入ArrayList对象customCommandLines中
public static void main(final String[] args) { EnvironmentInformation.logEnvironmentInfo(LOG, "Command Line Client", args); ...... // 3. load the custom command lines final ListcustomCommandLines = loadCustomCommandLines( configuration, configurationDirectory); ...... }
在后面 run() -> validateAndGetActiveCommandLine()方法中依次从customCommandLines对象中取出命令行对象,调用isActive()方法,判断是哪一种命令行
public CustomCommandLine validateAndGetActiveCommandLine(CommandLine commandLine) { for (CustomCommandLine cli : customCommandLines) { if (cli.isActive(commandLine)) { return cli; } } throw new IllegalStateException("No valid command-line found."); }
FlinkYarnSessionCli的isActive()方法中,会去判断运行bin/flink脚本时是否传入了-m参数,其值是否为yarn-cluster
@Override public boolean isActive(CommandLine commandLine) { final String jobManagerOption = commandLine.getOptionValue(addressOption.getOpt(), null); final boolean yarnJobManager = ID.equals(jobManagerOption); final boolean hasYarnAppId = commandLine.hasOption(applicationId.getOpt()) || configuration.getOptional(YarnConfigOptions.APPLICATION_ID).isPresent(); final boolean hasYarnExecutor = YarnSessionClusterExecutor.NAME.equalsIgnoreCase(configuration.get(DeploymentOptions.TARGET)) || YarnJobClusterExecutor.NAME.equalsIgnoreCase(configuration.get(DeploymentOptions.TARGET)); return hasYarnExecutor || yarnJobManager || hasYarnAppId || (isYarnPropertiesFileMode(commandLine) && yarnApplicationIdFromYarnProperties != null); }
addressOption为匹配-m,ID为yarn-cluster
(2)在CliFrontend的run()方法中,通过getEffectiveConfiguration()方法得到Configuration对象,传入的命令行对象activeCommandLine即为上面第一个步骤中得到的FlinkYarnSessionCli;
在getEffectiveConfiguration()方法中会调用FlinkYarnSessionCli的applyCommandLineOptionsToConfiguration()方法来增加和yarn相关的配置,代码如下:
public Configuration applyCommandLineOptionsToConfiguration(CommandLine commandLine) throws FlinkException { // we ignore the addressOption because it can only contain "yarn-cluster" final Configuration effectiveConfiguration = new Configuration(configuration); applyDescriptorOptionToConfig(commandLine, effectiveConfiguration); final ApplicationId applicationId = getApplicationId(commandLine); if (applicationId != null) { final String zooKeeperNamespace; if (commandLine.hasOption(zookeeperNamespace.getOpt())){ zooKeeperNamespace = commandLine.getOptionValue(zookeeperNamespace.getOpt()); } else { zooKeeperNamespace = effectiveConfiguration.getString(HA_CLUSTER_ID, applicationId.toString()); } effectiveConfiguration.setString(HA_CLUSTER_ID, zooKeeperNamespace); effectiveConfiguration.setString(YarnConfigOptions.APPLICATION_ID, ConverterUtils.toString(applicationId)); effectiveConfiguration.setString(DeploymentOptions.TARGET, YarnSessionClusterExecutor.NAME); } else { effectiveConfiguration.setString(DeploymentOptions.TARGET, YarnJobClusterExecutor.NAME); } ...... ...... }
其中关键配置DeploymentOptions.TARGET,即程序目标运行环境;YarnJobClusterExecutor.NAME 值为
public enum YarnDeploymentTarget { PER_JOB("yarn-per-job"), ..... }
在ClientUitls的executeProgram()中通过下面代码设置:
public static void executeProgram( PipelineExecutorServiceLoader executorServiceLoader, Configuration configuration, PackagedProgram program, boolean enforceSingleJobExecution, boolean suppressSysout) throws ProgramInvocationException { ...... try { ...... StreamContextEnvironment.setAsContext( executorServiceLoader, configuration, userCodeClassLoader, enforceSingleJobExecution, suppressSysout); ...... } finally { Thread.currentThread().setContextClassLoader(contextClassLoader); } }
ClientUtils的executeProgram()方法中调用PackagedProgram的invokeInteractiveModeForExecution(),来执行用户main()方法
private static void callMainMethod(Class> entryClass, String[] args) throws ProgramInvocationException { Method mainMethod; if (!Modifier.isPublic(entryClass.getModifiers())) { ...... } try { mainMethod = entryClass.getMethod("main", String[].class); } catch (NoSuchMethodException e) { ...... } catch (Throwable t) { ...... } if (!Modifier.isStatic(mainMethod.getModifiers())) { ...... } if (!Modifier.isPublic(mainMethod.getModifiers())) { ...... } try { mainMethod.invoke(null, (Object) args); } catch (IllegalArgumentException e) { ...... } catch (IllegalAccessException e) { ...... } catch (InvocationTargetException e) { ...... } catch (Throwable t) { } }
到此,相信大家对“Flink怎么执行用户程序”有了更深的了解,不妨来实际操作一番吧!这里是创新互联网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!