本篇内容主要讲解“Flink怎么执行用户程序”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“Flink怎么执行用户程序”吧!
以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 List<CustomCommandLine> customCommandLines = 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怎么执行用户程序”有了更深的了解，不妨来实际操作一番吧！这里是亿速云网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！
亿速云「云服务器」，即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘，价格低至29元/月。点击查看>>
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
原文链接：https://my.oschina.net/catcloud/blog/5001050
