Flink區分運行環境的方法是什么

這篇文章主要介紹“Flink區分運行環境的方法是什么”，在日常操作中，相信很多人在Flink區分運行環境的方法是什么問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”Flink區分運行環境的方法是什么”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

Flink判斷運行環境（本地、集群）的邏輯如下：

（1）在任務的main方法中，通過 StreamExecutionEnvironment 獲取運行環境

StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

（2）生成運行環境的工廠類放在ThreadLocal中；threadLocalContextEnvironmentFactory 是StreamExecutionEnvironment類的靜態屬性 

	/** The ThreadLocal used to store {@link StreamExecutionEnvironmentFactory}. */
	private static final ThreadLocal<StreamExecutionEnvironmentFactory> threadLocalContextEnvironmentFactory = new ThreadLocal<>();

①當是本地IDE直接運行任務main方法時，ThreadLocal中獲取到的StreamExecutionEnvironmentFactory為空，此時生成本地運行環境LocalStreamEnvironment

	public static StreamExecutionEnvironment getExecutionEnvironment() {
		return Utils.resolveFactory(threadLocalContextEnvironmentFactory, contextEnvironmentFactory)
			.map(StreamExecutionEnvironmentFactory::createExecutionEnvironment)
			.orElseGet(StreamExecutionEnvironment::createLocalEnvironment);
	}

當ThreadLocal中有StreamExecutionEnvironmentFactory時，則用其createExecutionEnvironment()方法來生成運行環境

②當集群環境時，是如何將StreamExecutionEnvironmentFactory放入到ThreadLocal中？

通過 bin/flink run ....   命令提交jar包到集群運行命令時，該腳本會調用 org.apache.flink.client.cli.CliFrontend  來運行用戶程序，如下：

.......
.......
# Add HADOOP_CLASSPATH to allow the usage of Hadoop file systems
exec $JAVA_RUN $JVM_ARGS $FLINK_ENV_JAVA_OPTS "${log_setting[@]}" -classpath "`manglePathList "$CC_CLASSPATH:$INTERNAL_HADOOP_CLASSPATHS"`" org.apache.flink.client.cli.CliFrontend "$@"

在CliFrontend中依次執行以下方法 main() ->  parseParameters() -> run() -> executeProgram() 

	protected void executeProgram(final Configuration configuration, final PackagedProgram program) throws ProgramInvocationException {
		ClientUtils.executeProgram(new DefaultExecutorServiceLoader(), configuration, program, false, false);
	}

在org.apache.flink.client.ClientUtils的executeProgram()中調用 StreamContextEnvironment.setAsContext(...)，StreamContextEnvironment繼承自StreamExecutionEnvironment。setAsContext()代碼如下

	public static void setAsContext(
			final PipelineExecutorServiceLoader executorServiceLoader,
			final Configuration configuration,
			final ClassLoader userCodeClassLoader,
			final boolean enforceSingleJobExecution,
			final boolean suppressSysout) {
		StreamExecutionEnvironmentFactory factory = () -> new StreamContextEnvironment(
			executorServiceLoader,
			configuration,
			userCodeClassLoader,
			enforceSingleJobExecution,
			suppressSysout);
		initializeContextEnvironment(factory);
	}

創建生成運行環境的工廠類實例，在initializeContextEnvironment()方法中把實例放到StreamExecutionEnvironment類的靜態屬性threadLocalContextEnvironmentFactory 中 ，代碼如下

	protected static void initializeContextEnvironment(StreamExecutionEnvironmentFactory ctx) {
		contextEnvironmentFactory = ctx;
		threadLocalContextEnvironmentFactory.set(contextEnvironmentFactory);
	}

這樣在用戶程序 StreamExecutionEnvironment.getExecutionEnvironment() 時，獲取到的運行環境就是StreamContextEnvironment類的setAsContext()方法中生成的

	public static void setAsContext(
			final PipelineExecutorServiceLoader executorServiceLoader,
			final Configuration configuration,
			final ClassLoader userCodeClassLoader,
			final boolean enforceSingleJobExecution,
			final boolean suppressSysout) {
		StreamExecutionEnvironmentFactory factory = () -> new StreamContextEnvironment(
			executorServiceLoader,
			configuration,
			userCodeClassLoader,
			enforceSingleJobExecution,
			suppressSysout);
		......
	}

本地運行環境LocalStreamEnvironment 和 獨立集群、flink on yarn等運行環境StreamContextEnvironment 的主要區別在于，他們的成員屬性 configuration 不同。LocalStreamEnvironment 中是創建的空鍵值對(new Configuration())，而StreamContextEnvironment 是通過 CliFrontend 生成的 Configuration 對象。

到此，關于“Flink區分運行環境的方法是什么”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續學習更多相關知識，請繼續關注億速云網站，小編會繼續努力為大家帶來更多實用的文章！