大數據流處理批處理數據工程【免費下載鏈接】flink項目地址https://gitcode.com/gh_mirrors/fli/flink點擊查看免費下載在 Flink 中無論是批處理還是流處理應用幾乎都依賴外部配置參數來驅動運行它們用于指定輸入輸出源如路徑或地址、系統(tǒng)參數并行度、運行時配置以及應用特有參數通常在用戶函數內部使用。本文圍繞 docs/content/docs/dev/datastream/application_parameters.md 的核心內容深入講解 Flink 提供的ParameterTool工具如何從.properties文件、命令行參數、系統(tǒng)屬性等多種來源加載配置如何在 DataStream/DataSet 程序中直接讀取參數、設置算子并行度、把參數傳入用戶函數以及如何將參數注冊為全局作業(yè)參數以便在算子函數與 Web 界面中訪問。讀完本文你將掌握一套可復制、可落地的 Flink 作業(yè)參數化實戰(zhàn)方案并理解其底層實現原理。一、ParameterTool 是什么ParameterTool是 Flink 自帶的一個簡單實用的參數讀取與解析工具類用于解決如何把外部配置送入 Flink 程序這一普遍問題。它的實現位于 ParameterTool.java并被標注為Public公共穩(wěn)定 API其類注釋明確指出This class provides simple utility methods for reading and parsing program arguments from different sources. Only single value parameter could be supported in args.也就是說ParameterTool內部本質上期望一個MapString, String因此非常容易與你自己的配置風格配置文件、環(huán)境變量、啟動參數等集成。需要說明的是你并不強制要求使用ParameterTool。其他框架如 Apache Commons CLI、argparse4j 等同樣可以與 Flink 配合得很好。ParameterTool的價值在于開箱即用、零額外依賴并且專門針對 Flink 的常見使用場景做了設計例如可序列化、可作為全局作業(yè)參數分發(fā)到所有算子。二、將配置值載入 ParameterToolParameterTool提供了一組預定義的靜態(tài)工廠方法用于從不同來源讀取配置。核心工廠方法都定義在 ParameterTool.java 中fromArgs(String[] args)從命令行參數解析fromPropertiesFile(String path)/fromPropertiesFile(File file)/fromPropertiesFile(InputStream inputStream)從.properties文件讀取fromSystemProperties()從 JVM 系統(tǒng)屬性讀取fromMap(MapString, String map)從任意 Map 構造2.1 從.properties文件讀取以下方法讀取 JavaProperties文件并提供鍵值對。三種重載形式分別接受文件路徑字符串、File對象和InputStreamString propertiesFilePath /home/sam/flink/myjob.properties; ParameterTool parameters ParameterTool.fromPropertiesFile(propertiesFilePath); File propertiesFile new File(propertiesFilePath); ParameterTool parameters ParameterTool.fromPropertiesFile(propertiesFile); InputStream propertiesFileInputStream new FileInputStream(file); ParameterTool parameters ParameterTool.fromPropertiesFile(propertiesFileInputStream);從源碼看fromPropertiesFile(String path)內部會先構造File再委托給fromPropertiesFile(File)而fromPropertiesFile(File)會先檢查文件是否存在不存在則拋出FileNotFoundException隨后通過FileInputStream加載并最終把Properties轉為 Map 構造ParameterTool。一個典型的myjob.properties內容形如inputhdfs:///mydata outputhdfs:///result expectedCount1000 mapParallelism42.2 從命令行參數解析這是最常用的方式支持類似--input hdfs:///mydata --elements 42的寫法public static void main(String[] args) { ParameterTool parameters ParameterTool.fromArgs(args); // .. regular code .. }fromArgs的解析規(guī)則見 ParameterTool.java值得深入了解鍵必須以-或--開頭后面跟隨值例如--key1 value1 --key2 value2 -key3 value3解析是順序掃描的遇到一個以-/--開頭的 token 記為 key然后看下一個 token若下一個 token 是數字通過NumberUtils.isNumber判斷則作為該 key 的值因此負數-0.58也能被正確識別為數值而非新的參數名若下一個 token 以-/--開頭說明它是另一個參數名則當前 key 被標記為無值參數存入常量NO_VALUE_KEY__NO_VALUE_KEY否則把下一個 token 作為值若 key 已經是最后一個 token同樣標記為NO_VALUE_KEY?？諈得鹝ey 為空字符串會拋出IllegalArgumentException。也就是說--flag無值和-Dxxx風格的參數都能被兼容處理之后可以用parameters.has(flag)判斷該參數是否存在。2.3 從系統(tǒng)屬性讀取啟動 JVM 時可以通過-Dinputhdfs:///mydata傳入系統(tǒng)屬性ParameterTool也支持直接從系統(tǒng)屬性初始化ParameterTool parameters ParameterTool.fromSystemProperties();其實現是fromMap((Map) System.getProperties())即把 JVM 的全部系統(tǒng)屬性包括java.version、os.name等都裝入ParameterTool。注意這意味著getNumberOfParameters()會包含所有 JVM 系統(tǒng)屬性實際使用時通常配合mergeWith或按需取值。2.4 從任意 Map 構造ParameterTool parameters ParameterTool.fromMap(map);fromMap會做非空校驗Preconditions.checkNotNull并基于傳入 Map 構造不可變副本Collections.unmodifiableMap。這使你可以方便地與自己的配置體系對接例如從環(huán)境變量構造、從 YAML/JSON 解析后的 Map 構造等。三、在 Flink 程序中讀取參數載入參數后有多種使用方式。3.1 直接從 ParameterTool 取值ParameterTool繼承自 AbstractParameterTool.java本身提供了豐富的取值方法ParameterTool parameters // ... parameters.getRequired(input); parameters.get(output, myDefaultValue); parameters.getLong(expectedCount, -1L); parameters.getNumberOfParameters(); // .. there are more methods available.完整的方法族包括方法行為get(String key)返回字符串值key 不存在時返回nullgetRequired(String key)key 不存在或值為空時拋出RuntimeExceptionNo data for required key ...get(String key, String defaultValue)不存在時返回默認值getInt/getLong/getFloat/getDouble/getBoolean/getShort/getByte每種類型都提供必填和帶默認值兩個重載值無法按對應類型解析時拋異常has(String key)判斷 key 是否存在getNumberOfParameters()返回參數總數從源碼看帶默認值的方法會先把默認值記錄到內部defaultData通過addToDefaults這對后續(xù)生成 properties 骨架文件createPropertiesFile有直接作用getRequired內部先調用get若返回null則拋出RuntimeException。此外AbstractParameterTool還提供了getUnrequestedParameters()返回尚未被get/has請求過的參數名集合——可用于校驗用戶傳了但程序沒用到的參數幫助排查拼寫錯誤。3.2 在 main() 中直接使用示例設置算子并行度你可以直接在提交應用的客戶端main()方法中使用這些方法的返回值。例如根據命令行參數設置算子并行度ParameterTool parameters ParameterTool.fromArgs(args); int parallelism parameters.get(mapParallelism, 2); DataStreamTuple2String, Integer counts text.flatMap(new Tokenizer()).setParallelism(parallelism);這里get(mapParallelism, 2)意味著命令行若提供--mapParallelism 8則并行度為 8否則回退到默認值 2。3.3 將 ParameterTool 傳給用戶函數由于ParameterTool實現了Serializable其父類AbstractParameterTool同時實現了ExecutionConfig.GlobalJobParameters與Serializable可以直接把它作為構造參數傳入函數隨算子一起被序列化分發(fā)到集群ParameterTool parameters ParameterTool.fromArgs(args); DataStreamTuple2String, Integer counts text.flatMap(new Tokenizer(parameters));之后在函數內部直接使用命令行讀取到的值public static final class Tokenizer extends RichFlatMapFunctionString, Tuple2String, Integer { private final ParameterTool parameters; public Tokenizer(ParameterTool parameters) { this.parameters parameters; } Override public void flatMap(String value, CollectorTuple2String, Integer out) { String input parameters.getRequired(input); // .. do more .. } }3.4 全局注冊參數Global Job Parameters將參數注冊為ExecutionConfig上的全局作業(yè)參數后它們會作為配置值出現在JobManager Web 界面中并且可以在所有用戶自定義函數里被訪問。注冊方式ParameterTool parameters ParameterTool.fromArgs(args); // set up the execution environment final ExecutionEnvironment env ExecutionEnvironment.getExecutionEnvironment(); env.getConfig().setGlobalJobParameters(parameters);對應的實現位于 ExecutionConfig.javasetGlobalJobParameters(GlobalJobParameters)會把參數轉換為MapString, String并內部存儲getGlobalJobParameters()返回存儲的實例。ParameterTool正是通過繼承ExecutionConfig.GlobalJobParameters并實現其抽象方法toMap()來無縫接入這套機制的見 AbstractParameterTool.java。在任意富函數Rich Function中訪問全局參數public static final class Tokenizer extends RichFlatMapFunctionString, Tuple2String, Integer { Override public void flatMap(String value, CollectorTuple2String, Integer out) { ParameterTool parameters ParameterTool.fromMap(getRuntimeContext().getGlobalJobParameters()); parameters.getRequired(input); // .. do more .. } }這里getRuntimeContext().getGlobalJobParameters()返回的正是ExecutionConfig.GlobalJobParameters由于ParameterTool繼承自它類型強轉或在運行時實際就是ParameterTool再用ParameterTool.fromMap(...)包一層即可復用全部取值方法。這種方式避免了在每個函數里手動傳遞參數對象的繁瑣且富函數天然擁有getRuntimeContext()訪問能力是最推薦的做法。四、源碼級進階ParameterTool 的更多能力圍繞ParameterTool倉庫中還提供了若干進階能力可以顯著提升實際工程中的可用性。4.1 MultipleParameterTool支持一個鍵對應多個值MultipleParameterTool.java標注為PublicEvolving是ParameterTool的多值版本用于處理形如--multi multiValue1 --multi multiValue2的重復參數。其類注釋明確指出Multiple values parameter in args could be supported. For example, --multi multiValue1 --multi multiValue2. If MultipleParameterTool object is used for GlobalJobParameters, the last one of multiple values will be used.使用方式MultipleParameterTool parameters MultipleParameterTool.fromArgs(args); // 獲取某個 key 的全部值 CollectionString multiValues parameters.getMultiParameter(multi); // 必填版 CollectionString required parameters.getMultiParameterRequired(multi); // 單值語義內部校驗必須恰好一個值否則拋異常 String single parameters.get(input);注意當MultipleParameterTool被用作GlobalJobParameters時其toMap()實現getFlatMapOfData會取多值中的最后一個作為該 key 的值這與單值ParameterTool的行為保持一致。4.2 mergeWith合并多個參數來源ParameterTool fromCli ParameterTool.fromArgs(args); ParameterTool fromProps ParameterTool.fromPropertiesFile(propertiesFilePath); ParameterTool merged fromCli.mergeWith(fromProps);mergeWith會把兩個ParameterTool的鍵值合并成一個新的ParameterTool后者覆蓋前者同名鍵同時正確合并已被請求的參數追蹤狀態(tài)。常見的組合用法是以命令行參數覆蓋配置文件默認值實現分層配置。4.3 導出與配置文件骨架生成// 轉成 Flink Configuration Configuration conf parameters.getConfiguration(); // 轉成 Properties Properties props parameters.getProperties(); // 生成 properties 骨架文件基于所有被 get/has 請求過的 key 及其默認值 parameters.createPropertiesFile(/path/to/default.properties, true);createPropertiesFile是很有用的配套工具它會把所有調用過get*/has的 key連同默認值未定義默認值的標記為undefined寫成一個 properties 文件方便你為作業(yè)快速生成配置模板。第二個參數overwrite控制是否允許覆蓋已存在文件為false時若文件已存在會拋出RuntimeException。4.4 序列化與并發(fā)安全ParameterTool實現了自定義的readObject反序列化邏輯會在反序列化時重建defaultDataConcurrentHashMap與unrequestedParameters并發(fā)安全的 Set確保對象在算子間傳輸后仍可正常工作。測試 ParameterToolTest.java 中的testConcurrentExecutionConfigSerialization對應 FLINK-7943驗證了并發(fā)序列化與并發(fā)訪問場景下的正確性。五、測試用例驗證解析行為一覽倉庫中的單元測試 ParameterToolTest.java 從多個角度驗證了上述行為可幫助你建立對解析規(guī)則的精確認知testFromCliArgs驗證命令行解析包括--input myInput標準鍵值、-expectedCount 15單橫線鍵、--withoutValues無值參數has(withoutValues) true、負數值-0.58能被正確解析為-expectedCount之外的新鍵的浮點值、布爾值true、字節(jié)與短整型邊界值等共解析出 7 個參數testFromPropertiesFile分別通過File與InputStream兩種方式從 properties 文件加載并校驗testFromMapOrProperties驗證從Properties/Map構造testSystemProperties驗證-D系統(tǒng)屬性方式加載testMerged驗證mergeWith合并命令行參數與系統(tǒng)屬性。這些測試共同構成了ParameterTool行為的事實依據閱讀它們可以快速理解各種邊界情況負數、無值參數、多來源合并等的實際表現。六、最佳實踐小結基于文檔與源碼在實際 Flink 作業(yè)中建議按如下模式組織參數處理統(tǒng)一入口在main()方法中集中加載參數推薦優(yōu)先級為命令行參數 properties 文件 代碼內默認值用mergeWith實現分層覆蓋盡早校驗對必需參數使用getRequired讓作業(yè)在提交階段快速失敗而不是在運行很久后才發(fā)現配置缺失全局注冊通過env.getConfig().setGlobalJobParameters(parameters)注冊全局參數在富函數中經getRuntimeContext().getGlobalJobParameters()獲取避免層層手動傳參同時還能在 JobManager Web 界面上直接查看作業(yè)參數便于排查問題類型化取值優(yōu)先使用getInt/getLong/getBoolean等類型化方法避免手工字符串轉換與解析錯誤利用骨架生成用createPropertiesFile為作業(yè)生成配置模板降低新環(huán)境部署的配置成本多值場景需要重復參數如多個 topic 列表時改用MultipleParameterTool。通過ParameterToolFlink 應用的配置管理可以從散落的硬編碼與手工解析收斂為單一入口、類型安全、可追蹤、可全局訪問的標準化方案這也是其在 DataStream 與 DataSet 作業(yè)中被廣泛采用的根本原因。相關源碼與測試可進一步參閱 ParameterTool.java、AbstractParameterTool.java、MultipleParameterTool.java 與 ParameterToolTest.java。贊分享大數據流處理批處理數據工程【免費下載鏈接】flink項目地址https://gitcode.com/gh_mirrors/fli/flink點擊查看免費下載相關推薦Flink 應用程序參數處理完全指南使用 ParameterTool 管理外部配置Flink 應用程序參數處理完全指南使用 ParameterTool 管理外部配置 導讀 幾乎所有的批處理和流處理 Flink 應用程序都依賴外部配置參數來大數據流處理批處理數據工程終極XSStrike交互式配置指南如何通過prompt.py實現靈活參數設置終極XSStrike交互式配置指南如何通過prompt.py實現靈活參數設置 XSStrike是一款強大的XSS漏洞檢測工具而其核心模塊 core/prom滲透測試應用安全WaveInApp核心組件解析深入理解GLAudioVisualizationView工作原理WaveInApp核心組件解析深入理解GLAudioVisualizationView工作原理 WaveInApp是一個強大的Android音頻可視化庫它能創(chuàng)作聲明：本文部分內容由AI輔助生成（AIGC），僅供參考