Flink怎么执行用户程序

本篇内容主要讲解“Flink怎么执行用户程序”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“Flink怎么执行用户程序”吧!

执行用户程序

CliFrontend生成Configuration对象

以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"),
    .....
}

给StreamExecutionEnvironment设置Configuration对象

在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);
		}
	}

运行用户程序main()方法

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怎么执行用户程序”有了更深的了解，不妨来实际操作一番吧！这里是亿速云网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！