您好,登錄后才能下訂單哦!
這篇文章主要介紹“Flink區(qū)分運行環(huán)境的方法是什么”,在日常操作中,相信很多人在Flink區(qū)分運行環(huán)境的方法是什么問題上存在疑惑,小編查閱了各式資料,整理出簡單好用的操作方法,希望對大家解答”Flink區(qū)分運行環(huán)境的方法是什么”的疑惑有所幫助!接下來,請跟著小編一起來學習吧!
Flink判斷運行環(huán)境(本地、集群)的邏輯如下:
(1)在任務的main方法中,通過 StreamExecutionEnvironment 獲取運行環(huán)境
StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
(2)生成運行環(huán)境的工廠類放在ThreadLocal中;threadLocalContextEnvironmentFactory 是StreamExecutionEnvironment類的靜態(tài)屬性
/** The ThreadLocal used to store {@link StreamExecutionEnvironmentFactory}. */ private static final ThreadLocal<StreamExecutionEnvironmentFactory> threadLocalContextEnvironmentFactory = new ThreadLocal<>();
①當是本地IDE直接運行任務main方法時,ThreadLocal中獲取到的StreamExecutionEnvironmentFactory為空,此時生成本地運行環(huán)境LocalStreamEnvironment
public static StreamExecutionEnvironment getExecutionEnvironment() { return Utils.resolveFactory(threadLocalContextEnvironmentFactory, contextEnvironmentFactory) .map(StreamExecutionEnvironmentFactory::createExecutionEnvironment) .orElseGet(StreamExecutionEnvironment::createLocalEnvironment); }
當ThreadLocal中有StreamExecutionEnvironmentFactory時,則用其createExecutionEnvironment()方法來生成運行環(huán)境
②當集群環(huán)境時,是如何將StreamExecutionEnvironmentFactory放入到ThreadLocal中?
通過 bin/flink run .... 命令提交jar包到集群運行命令時,該腳本會調(diào)用 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中依次執(zhí)行以下方法 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()中調(diào)用 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); }
創(chuàng)建生成運行環(huán)境的工廠類實例,在initializeContextEnvironment()方法中把實例放到StreamExecutionEnvironment類的靜態(tài)屬性threadLocalContextEnvironmentFactory 中 ,代碼如下
protected static void initializeContextEnvironment(StreamExecutionEnvironmentFactory ctx) { contextEnvironmentFactory = ctx; threadLocalContextEnvironmentFactory.set(contextEnvironmentFactory); }
這樣在用戶程序 StreamExecutionEnvironment.getExecutionEnvironment() 時,獲取到的運行環(huán)境就是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); ...... }
本地運行環(huán)境LocalStreamEnvironment 和 獨立集群、flink on yarn等運行環(huán)境StreamContextEnvironment 的主要區(qū)別在于,他們的成員屬性 configuration 不同。LocalStreamEnvironment 中是創(chuàng)建的空鍵值對(new Configuration()),而StreamContextEnvironment 是通過 CliFrontend 生成的 Configuration 對象。
到此,關于“Flink區(qū)分運行環(huán)境的方法是什么”的學習就結(jié)束了,希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習,快去試試吧!若想繼續(xù)學習更多相關知識,請繼續(xù)關注億速云網(wǎng)站,小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>
免責聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。