大數(shù)據(jù)流處理批處理數(shù)據(jù)工程【免費(fèi)下載鏈接】flink項(xiàng)目地址https://gitcode.com/gh_mirrors/fli/flink點(diǎn)擊查看免費(fèi)下載導(dǎo)讀本文是 Flink 項(xiàng)目中 PyFlink Table API 用戶自定義函數(shù)User Defined Functions的權(quán)威技術(shù)指南以倉庫內(nèi)官方參考文檔 flink-python/docs/reference/pyflink.table/udf.rst 為主體骨架結(jié)合 flink-python/pyflink/table/udf.py 源碼實(shí)現(xiàn)與 flink-python/pyflink/table/tests 下真實(shí)測試用例深度展開。讀完本文你將掌握標(biāo)量函數(shù)ScalarFunction/udf、表函數(shù)TableFunction/udtf、聚合函數(shù)AggregateFunction/udaf、表聚合函數(shù)TableAggregateFunction/udtaf四類 UDF 的定義方式、生命周期鉤子、類型聲明規(guī)則以及用ListView/MapView優(yōu)化聚合狀態(tài)性能的實(shí)戰(zhàn)技巧。一、PyFlink 用戶自定義函數(shù)體系總覽PyFlink 的 Table API 允許用戶在 SQL 與 Table 編程中使用 Python 編寫自定義函數(shù)官方將 UDF 劃分為四種語義類型每一類都在 flink-python/pyflink/table/udf.py 中提供了抽象基類與裝飾器/工廠函數(shù)兩種使用形態(tài)UDF 類型輸入輸出抽象基類創(chuàng)建助手標(biāo)量函數(shù)Scalar Function零個、一個或多個標(biāo)量值一個新的標(biāo)量值ScalarFunctionudf表函數(shù)Table Function零個、一個或多個標(biāo)量值零個、一個或多個行可結(jié)構(gòu)化類型TableFunctionudtf聚合函數(shù)Aggregate Function多行的標(biāo)量值一個新的標(biāo)量值A(chǔ)ggregateFunctionudaf表聚合函數(shù)Table Aggregate Function多行的標(biāo)量值零個、一個或多個行或結(jié)構(gòu)化類型TableAggregateFunctionudtaf四類函數(shù)分別對應(yīng) udf.py 中的ScalarFunction第 105 行、TableFunction第 121 行、AggregateFunction第 214 行、TableAggregateFunction第 235 行四個抽象類以及udf第 627 行、udtf第 688 行、udaf第 741 行、udtaf第 789 行四個創(chuàng)建助手。所有用戶自定義函數(shù)都繼承自公共基類UserDefinedFunctionudf.py它定義了三個貫穿函數(shù)整個生命周期的默認(rèn)鉤子open(function_context)初始化方法在主工作方法之前被調(diào)用適合做一次性設(shè)置如建立連接、加載模型。close()收尾方法在最后一次調(diào)用主工作方法后被調(diào)用用于釋放資源。is_deterministic()返回函數(shù)結(jié)果的確定性。默認(rèn)返回True只有當(dāng)調(diào)用保證在給定相同參數(shù)時永遠(yuǎn)返回相同結(jié)果時才應(yīng)為真像random()、date()、now()這類非純函數(shù)必須返回False。此外 udf.py 中的FunctionContext類用于在函數(shù)內(nèi)部獲取運(yùn)行時上下文信息提供兩個方法get_metric_group()返回當(dāng)前并行子任務(wù)的指標(biāo)組Metric Group未開啟指標(biāo)時拋出RuntimeError并提示通過python.metric.enabled配置開啟。get_job_parameter(key, default_value)按 key 獲取全局作業(yè)參數(shù)的值參數(shù)不存在時返回默認(rèn)值。二、標(biāo)量函數(shù)Scalar Function2.1 語義與抽象基類標(biāo)量函數(shù)將零個、一個或多個標(biāo)量值映射為一個新的標(biāo)量值?；怱calarFunction的核心是一個抽象方法eval(*args)它定義了標(biāo)量函數(shù)的計(jì)算邏輯udf.py。2.2 用udf裝飾器快速定義推薦使用udf助手函數(shù)創(chuàng)建標(biāo)量函數(shù)其完整簽名如下udf.pydef udf(fNone, input_typesNone, result_typeNone, deterministicNone, nameNone, func_typegeneral, udf_typeNone)參數(shù)說明參數(shù)類型說明f可調(diào)用對象 /ScalarFunction實(shí)例 / 類函數(shù)體可為 lambda、普通 Python 函數(shù)、functools.partial、實(shí)現(xiàn)了__call__的可調(diào)用對象或ScalarFunction子類實(shí)例input_types可選輸入數(shù)據(jù)類型DataType、strSQL 類型名或其列表也可傳RowTyperesult_type必填結(jié)果數(shù)據(jù)類型DataType或strdeterministic可選默認(rèn)True函數(shù)結(jié)果的確定性name可選函數(shù)名默認(rèn)取函數(shù)__name__或類名func_typegeneral/pandas默認(rèn)generalPython 函數(shù)的執(zhí)行類型udf_type為 1.12 起廢棄的舊參數(shù)名會觸發(fā)DeprecationWarning三種等價的定義方式來自 udf.py 的官方示例from pyflink.table import DataTypes from pyflink.table.udf import ScalarFunction, udf # 方式一lambda 顯式 input_types add_one udf(lambda i: i 1, DataTypes.BIGINT(), DataTypes.BIGINT()) # 方式二裝飾器input_types 可選 udf(result_typeDataTypes.BIGINT()) def add(i, j): return i j # 方式三result_type 用字符串指定 SQL 類型 udf(result_typeBIGINT) def add(i, j): return i j2.3 繼承ScalarFunction實(shí)現(xiàn)有狀態(tài)邏輯當(dāng)需要狀態(tài)或資源時繼承ScalarFunction并實(shí)現(xiàn)eval、可選實(shí)現(xiàn)open/closefrom pyflink.table.udf import ScalarFunction, FunctionContext class SubtractOne(ScalarFunction): def eval(self, i): return i - 1 # 結(jié)合 FunctionContext 使用全局作業(yè)參數(shù)與指標(biāo) class SubtractWithParameters(ScalarFunction): def open(self, function_context: FunctionContext): self.subtract_value int(function_context.get_job_parameter(subtract_value, 0)) self.metric_group function_context.get_metric_group() self.counter self.metric_group.add_group(my_group, g).counter(my_counter) def eval(self, i): self.counter.inc() return i - self.subtract_value測試用例 flink-python/pyflink/table/tests/test_udf.py 在test_scalar_function第 40 行中驗(yàn)證了 lambda、ScalarFunction子類、可調(diào)用對象、functools.partial四種形態(tài)的標(biāo)量函數(shù)并通過python.metric.enabled與pipeline.global-job-parameters配置分別驗(yàn)證了指標(biāo)禁用與全局參數(shù)傳遞兩種場景。2.4 確定性、命名與數(shù)據(jù)類型覆蓋test_deterministictest_udf.py驗(yàn)證默認(rèn)確定性為True當(dāng)傳入deterministicFalse與基類is_deterministic()返回True沖突時構(gòu)造階段會拋出ValueError: Inconsistent deterministic。test_name第 744 行驗(yàn)證name參數(shù)可覆蓋默認(rèn)函數(shù)名。test_all_data_types第 499 行與test_all_data_types_expression第 284 行覆蓋了BOOLEAN/TINYINT/SMALLINT/INT/BIGINT/FLOAT/DOUBLE/BYTES/STRING/DATE/TIME/TIMESTAMP/ARRAY/MAP/DECIMAL/BINARY/CHAR/VARCHAR等全部內(nèi)置類型的輸入輸出映射。三、表函數(shù)Table Function3.1 語義與抽象基類表函數(shù)將零個、一個或多個標(biāo)量值映射為零行、一行或多行。基類TableFunction的核心同樣是抽象方法eval(*args)udf.py但語義上要求以迭代器generator方式產(chǎn)出多行結(jié)果。3.2 用udtf裝飾器定義udtf的完整簽名udf.pydef udtf(fNone, input_typesNone, result_typesNone, deterministicNone, nameNone)其中result_types支持三種指定方式官方示例見 udf.pyfrom pyflink.table import DataTypes from pyflink.table.udf import TableFunction, udtf # 方式一DataType 列表 udtf(result_types[DataTypes.BIGINT(), DataTypes.BIGINT()]) def range_emit(s, e): for i in range(e): yield s, i # 方式二SQL 類型名字符串列表 udtf(result_types[BIGINT, BIGINT]) def range_emit(s, e): for i in range(e): yield s, i # 方式三ROW 字符串命名列 udtf(result_typesRowa BIGINT, b BIGINT) def range_emit(s, e): for i in range(e): yield s, i # 方式四繼承 TableFunctioneval 返回可迭代對象 class MultiEmit(TableFunction): def eval(self, i): return range(i) multi_emit udtf(MultiEmit(), DataTypes.BIGINT(), DataTypes.BIGINT())從源碼看UserDefinedTableFunctionWrapper._create_judfudf.py會對result_types做歸一化字符串列表會被拼裝為Rowf0 ..., f1 ...行類型多個DataType會被包裝為DataTypes.ROW(...)最終映射到 Java 側(cè)的PythonTableFunction。3.3 表函數(shù)的使用與測試驗(yàn)證表函數(shù)在 SQL 中通過LATERAL TABLE(...)使用測試 flink-python/pyflink/table/tests/test_udtf.py 的test_table_function_with_sql_query第 54 行展示了注冊與調(diào)用self.t_env.create_temporary_system_function( multi_emit, udtf(MultiEmit(), result_types[DataTypes.BIGINT(), DataTypes.BIGINT()])) result self.t_env.sql_query(SELECT a, s, e FROM MyTable, LATERAL TABLE(multi_emit(a, a)))該測試同時覆蓋了流式PyFlinkStreamUserDefinedFunctionTests與批式PyFlinkBatchUserDefinedFunctionTests兩種執(zhí)行環(huán)境說明表函數(shù)在兩類模式下均可直接使用。四、聚合函數(shù)Aggregate Function4.1 命令式聚合的抽象契約聚合函數(shù)將多行的標(biāo)量值聚合成一個新的標(biāo)量值。它不再依賴單一的eval而是采用累加器accumulator模式抽象基類為ImperativeAggregateFunctionudf.pyAggregateFunction在其上增加抽象方法get_value(accumulator)第 222-232 行用于物化聚合結(jié)果。必須實(shí)現(xiàn)的方法方法職責(zé)create_accumulator()創(chuàng)建并初始化累加器返回初始值accumulate(accumulator, *args)處理輸入值并更新累加器實(shí)例get_value(accumulator)每次需要物化聚合結(jié)果時調(diào)用返回值可以是早到的、不完整的結(jié)果數(shù)據(jù)到達(dá)時周期性下發(fā)也可以是最終聚合結(jié)果可選方法retract(accumulator, *args)從累加器中撤回之前累計(jì)的輸入值用于處理流式場景中的更新與回撤如GROUP BY上的流聚合。默認(rèn)拋出RuntimeError(Method retract is not implemented)。merge(accumulator, accumulators)將一組累加器合并為一個。對于無界會話窗口分組聚合和有界分組聚合必須實(shí)現(xiàn)注意accumulator中可能已含有之前的聚合結(jié)果不能替換或清空該實(shí)例。get_result_type()/get_accumulator_type()返回結(jié)果與累加器的DataType實(shí)現(xiàn)類通常覆寫它們或在使用udaf時通過參數(shù)顯式指定。4.2 用udaf裝飾器定義udaf的完整簽名udf.pydef udaf(fNone, input_typesNone, result_typeNone, accumulator_typeNone, deterministicNone, nameNone, func_typegeneral)測試用例 flink-python/pyflink/table/tests/test_udaf.py 提供了多個可直接參考的完整實(shí)現(xiàn)例如CountAggregateFunction第 40-62 行from pyflink.table.udf import AggregateFunction, udaf class CountAggregateFunction(AggregateFunction): def get_value(self, accumulator): return accumulator[0] def create_accumulator(self): return [0] def accumulate(self, accumulator, *args): accumulator[0] accumulator[0] 1 def retract(self, accumulator, *args): accumulator[0] accumulator[0] - 1 def merge(self, accumulator, accumulators): for other_acc in accumulators: accumulator[0] accumulator[0] other_acc[0] def get_accumulator_type(self): return ARRAYBIGINT def get_result_type(self): return BIGINT my_count udaf(CountAggregateFunction())要點(diǎn)累加器類型與結(jié)果類型既可以覆寫get_accumulator_type()/get_result_type()提供也可以在udaf參數(shù)中顯式傳入兩者需同時為DataType或同時為str否則拋出TypeErrorudf.py。func_typepandas時udaf將 Python 函數(shù)包裝為 Pandas UDAF此時結(jié)果類型不能是MapType源碼中顯式拋出TypeError且累加器會被自動構(gòu)造為ARRAYresult_type。udaf(result_typeDataTypes.FLOAT(), func_typepandas) def mean_udaf(v): return v.mean()4.3 測試覆蓋的進(jìn)階特性test_udaf.py 驗(yàn)證了大量實(shí)戰(zhàn)場景test_double_aggregate第 254 行同一查詢中多個 UDAF 組合使用。test_mixed_with_built_in_functions_with_retract/without_retract第 277/326 行UDAF 與內(nèi)置函數(shù)混合以及是否實(shí)現(xiàn)retract對回撤路徑的影響。test_distinct_and_filter第 488 行UDAF 與DISTINCT、FILTER子句組合。test_tumbling_group_window_over_time/over_count、test_sliding_group_window_over_time/over_count、test_session_group_window_over_time第 547-770 行UDAF 在滾動/滑動/會話三種分組窗口上的正確性。test_clean_state第 518 行與test_data_view_clear第 425 行狀態(tài)清理與 DataView 的clear()語義。五、表聚合函數(shù)Table Aggregate Function5.1 語義與抽象方法表聚合函數(shù)將多行的標(biāo)量值映射為零行、一行或多行或結(jié)構(gòu)化類型。如果輸出記錄只包含一個字段可以省略結(jié)構(gòu)化記錄直接發(fā)射標(biāo)量值運(yùn)行時會自動將其包裝進(jìn)一行?；怲ableAggregateFunction在ImperativeAggregateFunction之上定義抽象方法emit_value(accumulator)udf.py它返回一個可迭代對象迭代器/生成器在每次需要物化聚合結(jié)果時被調(diào)用。5.2 用udtaf裝飾器定義udtaf的完整簽名udf.py與udaf一致但僅支持func_typegeneral傳入pandas會拋出ValueError。官方示例udf.py——一個經(jīng)典的 Top2 實(shí)現(xiàn)from pyflink.common import Row from pyflink.table.udf import TableAggregateFunction, udtaf class Top2(TableAggregateFunction): def emit_value(self, accumulator): yield Row(accumulator[0]) yield Row(accumulator[1]) def create_accumulator(self): return [None, None] def accumulate(self, accumulator, *args): if args[0] is not None: if accumulator[0] is None or args[0] accumulator[0]: accumulator[1] accumulator[0] accumulator[0] args[0] elif accumulator[1] is None or args[0] accumulator[1]: accumulator[1] args[0] def retract(self, accumulator, *args): accumulator[0] accumulator[0] - 1 def merge(self, accumulator, accumulators): for other_acc in accumulators: self.accumulate(accumulator, other_acc[0]) self.accumulate(accumulator, other_acc[1]) def get_accumulator_type(self): return ARRAYBIGINT def get_result_type(self): return ROWa BIGINT top2 udtaf(Top2())注意Top2中同時實(shí)現(xiàn)了retract與merge使其可用于流式回撤聚合與有界/會話窗口聚合場景。emit_value每次產(chǎn)出多行這正是表聚合與普通聚合的核心差異。六、DataView大數(shù)據(jù)量累加器的性能利器6.1 為什么需要 DataView官方文檔udf.rst 的 DataView 一節(jié)明確說明如果累加器需要存儲大量數(shù)據(jù)應(yīng)當(dāng)使用pyflink.table.ListView和pyflink.table.MapView替代 list 和 dict。這兩種數(shù)據(jù)結(jié)構(gòu)提供與 list/dict 類似的功能但通常具有更好的性能——它們借助 Flink 的狀態(tài)后端state backend避免了對每個元素都產(chǎn)生不必要的狀態(tài)訪問開銷。6.2 ListView 與 MapView 的 API定義在 flink-python/pyflink/table/data_view.pyListView第 43-102 行提供get()、add(value)、add_all(values)、clear()支持迭代__iter__可視為狀態(tài)后端的 list。MapView第 105-197 行提供get(key)、put(key, value)、put_all(dict)、remove(key)、contains(key)、items()、keys()、values()、is_empty()、clear()并支持[]下標(biāo)讀寫__getitem__/__setitem__/__delitem__與in判斷__contains__可視為狀態(tài)后端的 dict。兩者均繼承DataView抽象類并實(shí)現(xiàn)clear()data_view.py。6.3 在累加器中聲明并使用在累加器類型中通過DataTypes.LIST_VIEW(...)與DataTypes.MAP_VIEW(...)聲明。這兩個類型工廠定義于 flink-python/pyflink/table/types.py官方文檔強(qiáng)調(diào)它們只能用于聚合函數(shù)的累加器類型聲明。以測試 test_udaf.py 中的ListViewConcatAggregateFunction第 116-138 行為例——用ListView累積大量字符串后拼接from pyflink.common import Row from pyflink.table import DataTypes from pyflink.table.udf import AggregateFunction from pyflink.table.data_view import ListView class ListViewConcatAggregateFunction(AggregateFunction): def get_value(self, accumulator): return accumulator[1].join(accumulator[0]) def create_accumulator(self): return Row(ListView(), ) def accumulate(self, accumulator, *args): accumulator[1] args[1] accumulator[0].add(args[0]) def get_accumulator_type(self): return DataTypes.ROW([ DataTypes.FIELD(f0, DataTypes.LIST_VIEW(DataTypes.STRING())), DataTypes.FIELD(f1, DataTypes.BIGINT())]) def get_result_type(self): return DataTypes.STRING()再看CountDistinctAggregateFunction第 140-173 行用MapView實(shí)現(xiàn)高效的去重計(jì)數(shù)累加器類型聲明為ROWf0 MAPSTRING, STRING, f1 BIGINT并演示了is_empty()、下標(biāo)讀寫、put_all()、clear()的完整用法class CountDistinctAggregateFunction(AggregateFunction): def get_value(self, accumulator): return accumulator[1] def create_accumulator(self): return Row(MapView(), 0) def accumulate(self, accumulator, *args): input_str args[0] if accumulator[0].is_empty() or input_str not in accumulator[0] \ or accumulator[0][input_str] is None: accumulator[0][input_str] 1 accumulator[1] 1 else: accumulator[0][input_str] 1 if input_str clear: accumulator[0].clear() accumulator[1] 0 def retract(self, accumulator, *args): input_str args[0] if accumulator[0].is_empty() or input_str not in accumulator[0]: return accumulator[0].put_all({input_str: accumulator[0][input_str] - 1}) if accumulator[0][input_str] 0: accumulator[1] - 1 accumulator[0][input_str] None def get_accumulator_type(self): return ROWf0 MAPSTRING, STRING, f1 BIGINT def get_result_type(self): return BIGINT相關(guān)測試覆蓋test_list_view第 369 行、test_map_view第 391 行、test_data_view_clear第 425 行驗(yàn)證聚合狀態(tài)下調(diào)clear()后的行為、test_map_view_iterate第 442 行驗(yàn)證items()/keys()/values()迭代。七、源碼級運(yùn)行機(jī)制7.1 從 Python 函數(shù)到 Java UDF 的橋接所有udf/udtf/udaf/udtaf返回的對象都是UserDefinedFunctionWrapper的子類包裝器udf.py。其內(nèi)部機(jī)制輸入類型歸一化input_types支持RowType、單個DataType/str或列表統(tǒng)一轉(zhuǎn)為列表非法類型拋出TypeError第 336-349 行。確定性檢查若顯式傳入deterministic且與函數(shù)自身is_deterministic()沖突拋出ValueError第 356-359 行未顯式指定時默認(rèn)取函數(shù)自身判定純 Python 函數(shù)默認(rèn)True第 361-363 行。函數(shù)調(diào)用包裝器實(shí)現(xiàn)__call__內(nèi)部通過expr.call(self, *args)將函數(shù)調(diào)用轉(zhuǎn)化為 Table API 的Expression第 368-370 行因此 UDF 可以直接參與 Table 表達(dá)式運(yùn)算與 SQL 解析。alias()與行輸入模式alias(*names)為表函數(shù)結(jié)果列命名第 372-374 行_set_takes_row_as_input()標(biāo)記以整行作為輸入的模式。序列化與 Java 橋接_java_user_defined_function()第 380-411 行通過cloudpickle.dumps()序列化 Python 函數(shù)并根據(jù)func_type映射到 Java 側(cè)PythonFunctionKind.GENERAL或PANDAS最終構(gòu)造 Java 側(cè)的PythonScalarFunction/PythonTableFunction/PythonAggregateFunction/PythonTableAggregateFunction對應(yīng)org.apache.flink.table.functions.python包通過PythonEnv.ExecType.PROCESS指定以獨(dú)立進(jìn)程方式執(zhí)行 Python 代碼第 601-605 行。7.2 lambda 與普通函數(shù)的委托包裝當(dāng)f不是UserDefinedFunction實(shí)例如 lambda、普通函數(shù)時包裝器通過_create_delegate_function()生成內(nèi)部委托實(shí)現(xiàn)DelegatingScalarFunction第 258-268 行、DelegationTableFunction第 271-281 行、DelegatingPandasAggregateFunction第 284-300 行g(shù)et_value直接取累加器首個元素從而使裝飾器語法與類繼承語法在底層統(tǒng)一。八、函數(shù)注冊與使用自定義函數(shù)創(chuàng)建后可通過以下方式注冊到 Table 環(huán)境from pyflink.table import EnvironmentSettings, TableEnvironment env_settings EnvironmentSettings.in_streaming_mode() t_env TableEnvironment.create(env_settings) # 注冊為臨時系統(tǒng)函數(shù)SQL 可直接調(diào)用 t_env.create_temporary_system_function(add_one, add_one) # 或注冊為臨時目錄函數(shù) t_env.create_temporary_function(add_one, add_one) # SQL 中使用 t_env.sql_query(SELECT add_one(a) FROM MyTable) # Table API 表達(dá)式中使用 from pyflink.table import expressions as expr result t_env.from_path(MyTable).select(expr.call(add_one, expr.col(a)))測試 test_udf.py 的test_create_and_drop_function第 700 行演示了注冊、調(diào)用與drop_function刪除的完整生命周期test_overwrite_builtin_function第 233 行還展示了通過注冊同名 UDF 覆蓋內(nèi)置函數(shù)如重定義plus的能力。九、最佳實(shí)踐與常見陷阱必須指定result_type或覆寫get_result_typeudf的result_type是必填參數(shù)類型既可用DataTypes.xxx()也可用 SQL 類型字符串如BIGINT字符串形式在 udf.py 中被直接透傳給 Java 側(cè)解析。確定性聲明影響優(yōu)化非純函數(shù)務(wù)必設(shè)置deterministicFalse如隨機(jī)數(shù)、取當(dāng)前時間否則 Flink 可能基于錯誤假設(shè)做查詢優(yōu)化同時注意deterministic與is_deterministic()沖突會直接報錯。流式聚合實(shí)現(xiàn)retract默認(rèn)retract拋異常涉及回撤如GROUP BY流聚合、與內(nèi)置聚合函數(shù)混合時需實(shí)現(xiàn)參考 test_udaf.py 的混合場景測試。會話窗口/有界聚合實(shí)現(xiàn)merge無界會話窗口分組聚合與有界分組聚合要求實(shí)現(xiàn)merge且不能替換傳入的accumulator實(shí)例。大數(shù)據(jù)量累加器優(yōu)先用ListView/MapView按官方文檔建議替代裸 list/dict 以獲得狀態(tài)后端優(yōu)化DataTypes.LIST_VIEW/MAP_VIEW只能用于聚合函數(shù)累加器類型聲明。資源生命周期重資源連接、模型在open()中初始化、close()中釋放全局參數(shù)通過FunctionContext.get_job_parameter()讀取指標(biāo)需先開啟python.metric.enabled。十、參考文檔與源碼索引官方參考文檔flink-python/docs/reference/pyflink.table/udf.rst核心實(shí)現(xiàn)flink-python/pyflink/table/udf.py基類與裝飾器、flink-python/pyflink/table/data_view.pyDataView、flink-python/pyflink/table/types.pyLIST_VIEW/MAP_VIEW類型工廠測試用例flink-python/pyflink/table/tests/test_udf.py、test_udtf.py、test_udaf.py類型定義參考flink-python/docs/reference/pyflink.table/data_types.rst贊分享大數(shù)據(jù)流處理批處理數(shù)據(jù)工程【免費(fèi)下載鏈接】flink項(xiàng)目地址https://gitcode.com/gh_mirrors/fli/flink點(diǎn)擊查看免費(fèi)下載相關(guān)推薦Flink Python Table API 用戶自定義函數(shù)完全指南UDF、UDTF、UDAGG 與 UDTAGG 實(shí)戰(zhàn)Flink Python Table API 用戶自定義函數(shù)完全指南UDF、UDTF、UDAGG 與 UDTAGG 實(shí)戰(zhàn) 本指南圍繞 Flink PyFlin大數(shù)據(jù)流處理批處理數(shù)據(jù)工程Flink自定義函數(shù)開發(fā)UDF、UDAF、UDTF編寫與注冊完整指南Flink自定義函數(shù)開發(fā)UDF、UDAF、UDTF編寫與注冊完整指南 Apache Flink作為業(yè)界領(lǐng)先的流處理框架提供了強(qiáng)大的自定義函數(shù)功能允許開發(fā)者示例工程大數(shù)據(jù)PyFlink 用戶自定義函數(shù)UDF / UDTF / UDAGG / UDTAGG完整實(shí)戰(zhàn)指南從 Python Table API 到 SQL 的四種函數(shù)定義與調(diào)用PyFlink 用戶自定義函數(shù)UDF / UDTF / UDAGG / UDTAGG完整實(shí)戰(zhàn)指南從 Python Table API 到 SQL 的四種函大數(shù)據(jù)流處理批處理數(shù)據(jù)工程上一篇精通Google API Python Client從基礎(chǔ)認(rèn)證到高級批量請求處理完整指南下一篇Spray高級技巧繞過賬戶鎖定策略的10種方法創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考