FlinkSQL 动态加载 UDF 实现思路( 三 ) 导读：最近在对Flink进行平台化

完整代码public static void main(String[] args) throws Exception {//创建流运行时环境StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment();//采用BlinkPlannerEnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();//创建StreamTable环境StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings);bsEnv.setParallelism(1);// 动态加载String funJarPath = "UDF jar 在 OSS 中所在的 URL 路径";loadJar(new URL(funJarPath));Field configurationField = StreamExecutionEnvironment.class.getDeclaredField("configuration");configurationField.setAccessible(true);Configuration o = (Configuration)configurationField.get(bsEnv);Field confData = http://kandian.youth.cn/index/Configuration.class.getDeclaredField("confData");confData.setAccessible(true);Map temp = (Map)confData.get(o);List jarList = new ArrayList<>();jarList.add(funJarPath);temp.put("pipeline.classpaths",jarList);bsTableEnv.executeSql("CREATE FUNCTION returnSelf AS 'flinksql.function.udf.ReturnSelf'");bsTableEnv.executeSql("CREATE TABLE sourceTable (" +"f_sequence INT," +"f_random INT," +"f_random_str STRING," +"ts AS localtimestamp," +"WATERMARK FOR ts AS ts" +") WITH (" +"'connector' = 'datagen'," +"'rows-per-second'='5'," +"'fields.f_sequence.kind'='sequence'," +"'fields.f_sequence.start'='1'," +"'fields.f_sequence.end'='1000'," +"'fields.f_random.min'='1'," +"'fields.f_random.max'='1000'," +"'fields.f_random_str.length'='10'" +")");bsTableEnv.executeSql("CREATE TABLE sinktable (" +"f_random_str STRING" +") WITH (" +"'connector' = 'print'" +")");bsTableEnv.executeSql("insert into sinktable select returnSelf(f_random_str) from sourceTable");}//动态加载Jarpublic static void loadJar(URL jarUrl) {//从URLClassLoader类加载器中获取类的addURL方法Method method = null;try {method = URLClassLoader.class.getDeclaredMethod("addURL", URL.class);} catch (NoSuchMethodException | SecurityException e1) {e1.printStackTrace();}// 获取方法的访问权限boolean accessible = method.isAccessible();try {//修改访问权限为可写if (accessible == false) {method.setAccessible(true);}// 获取系统类加载器URLClassLoader classLoader = (URLClassLoader) ClassLoader.getSystemClassLoader();//jar路径加入到系统url路径里method.invoke(classLoader, jarUrl);} catch (Exception e) {e.printStackTrace();} finally {method.setAccessible(accessible);}}最后以上就是在 Flink on K8S 集群 Session 模式下，FlinkSQL 动态加载 Jar 的解决方案。由于 REST API 没有提供 -C 效果，自定义 Jar URL 没有分发到 TaskMananger ，导致 TaskMananger 没有进行类加载到其 JVM 中。通过在 Job 的 main 方法中增加动态加载方法及配置 pipeline.classpaths ，可以达到不改动 Flink 源码的情况下实现 -C 效果。以上方案刚实现不久，还不保证是否有其他未知的问题，如果有更好的解决方案或者该方案中存在错误或者疏漏也欢迎提出共同讨论。
感谢您的阅读，如果喜欢本文欢迎关注和转发，本头条号将坚持持续分享IT技术知识。对于文章内容有其他想法或意见建议等，欢迎提出共同讨论共同进步。

FlinkSQL 动态加载 UDF 实现思路( 三 )

推荐阅读

民众发布|【中山民众新时代文明实践】今天你参与垃圾分类了吗？跟着志愿者学一学

胖哥一叶知秋：一个补丁却费银三千两，到底是节俭还是浪费?，道光皇帝十分抠门

鬼娃恰吉|《鬼娃恰吉》获得第三季续订，恐怖娃娃Chucky的恐惧之旅继续展开

经络|经络不通的6种反应：这些部位“硬”百病缠身！趁着夏天，让它们软下来！

蠢萌小萝莉|叶一茜为老公庆生，寿星穿一件绿色T恤青春四射，卡通照俏皮可爱

共财|亲戚不共财，共财断往来

网友真实经历灵异事件你有没有过真实的灵异经历?

请教：有哪些不常见到，却意义深刻的成语、俗语

『Python』为什么迁移至 Python 3 这么难？

大浩爱说新鲜事儿▲1公里“烧”多少钱？一般人真开不起！，价值2000万的劳斯莱斯

宣太后和秦武王,宣太后和武则天谁更有能力-

贴吧游戏精选|暗黑破坏神第一名第三名至今还有人玩，90后玩过的单机游戏大盘点

此时进场还有机会吗？各大国际知名投行对A股的看法汇总

朴槿惠|听到同父异母的姐姐去世朴槿惠狱中沉默了

物业|南都物业：2020年上半年净利润约6800万元，同比增加21.34%

哪篇古文（如项脊轩志）或者古诗曾给予你感动和力量

『体坛新闻资讯』探讨与76人交易的可能性

厨房油烟清洗妙招厨房油烟清洗小妙招

『追忆汉唐』从五帝钱中去认识大清兴盛时期

家常菜|营养美味的家常菜，超级下饭开胃，低脂低热量，零负担不长胖