1 11 flink中的動態載入udf jar包

專案中想要把flink做到平台化,只需要編輯sql便能把任務跑起來,開發過程中遇到乙個問題,就是如何能夠自動的載入自定義的函式包,因為專案中已經把main打包成乙個通用的jar, 使用時只需要把sql資訊用引數形式傳入就可以. 但是如果sql中需要使用到udf,那麼就需要實現flink的動態載入jar

具體**例子如下

public static void main(final string args) throws exception 
//動態載入jar
public static void loadjar(url jarurl) catch (nosuchmethodexception | securityexception e1) 
// 獲取方法的訪問許可權
boolean accessible = method.isaccessible();
try 
// 獲取系統類載入器
urlclassloader classloader = (urlclassloader) classloader.getsystemclassloader();
//jar路徑加入到系統url路徑裡
method.invoke(classloader, jarurl);
} catch (exception e) finally 
}

首先是loadjar方法是需要,在idea中,只是用了loadjar是可以載入外部udf包的,正常執行,但是當我打包放到集群上,則會提是找不到對應的class,猜測是因為idea開發中,啟動的是乙個minicluster, 所以我在idea中啟動的時候, main呼叫loadjar方法, 把udf載入到minicluster jvm程序中, jm跟tm都是在同個jvm程序中執行的, 所以是可以正常啟動. 但是如果是在集群環境, 即使是local standalone , 執行jps命令可以發現是jm跟tm是不同的程序的

所以是main即使在jm中載入了,但是tm程序中沒有載入

後面發現flink命令列啟動的時候可以新增-c引數, 可以指定其他的jar檔案

執行命令

flink --help

所以可以使用如下命令, 我的jar是上傳到阿里雲的oss上使用的

flink run -c "" -c cn.xuying.flink.stream.******test /usr/local/soft/flink/flink-1.0-snapshot.jar

但是我們專案平台是呼叫了集群的restapi來管理job的,而flink的restapi對應的介面是沒有提供-c的之類的引數的, 也在flink中文郵件列表提了對應的問題,不過沒有乙個很好的答案

中途大佬的參與,發現了flink 客戶端跑flinkjob的一些細節,

checkout原始碼,install,然後可以修改測試**

找到測試的job打斷點

執行test方法,可以發現env的裡面的屬性中多了-c引數url

然後我們不使用clifrontendruntest, 而是直接的執行那個testjob的main函式, 斷點可以發現沒有了-c的配置

所以推測, flink 的client主要是解析你的命令列, 加一些引數配置, 然後執行你編寫的flink main函式中的**, 所以我們可以在我們自己的job**中通過反射強制加入我們自己需要-c的配置 , 如文章開頭的**例子所示

1 11 flink中的動態載入udf jar包

flink中的類載入器ClassLoader

Unity中的動態載入資源

Angular中的動態元件載入

1 11 flink中的動態載入udf jar包

flink中的類載入器ClassLoader

Unity中的動態載入資源

Angular中的動態元件載入

相關推薦