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怎么執行用戶程序”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！