Flink PyFlink Table 用户自定义函数（UDF/UDTF/UDAF/UDTAAF）完全指南

大数据流处理批处理数据工程【免费下载链接】flink项目地址https://gitcode.com/gh_mirrors/fli/flink点击查看免费下载导读本文是 Flink 项目中 PyFlink Table API 用户自定义函数User Defined Functions的权威技术指南以仓库内官方参考文档 flink-python/docs/reference/pyflink.table/udf.rst 为主体骨架结合 flink-python/pyflink/table/udf.py 源码实现与 flink-python/pyflink/table/tests 下真实测试用例深度展开。读完本文你将掌握标量函数ScalarFunction/udf、表函数TableFunction/udtf、聚合函数AggregateFunction/udaf、表聚合函数TableAggregateFunction/udtaf四类 UDF 的定义方式、生命周期钩子、类型声明规则以及用ListView/MapView优化聚合状态性能的实战技巧。一、PyFlink 用户自定义函数体系总览PyFlink 的 Table API 允许用户在 SQL 与 Table 编程中使用 Python 编写自定义函数官方将 UDF 划分为四种语义类型每一类都在 flink-python/pyflink/table/udf.py 中提供了抽象基类与装饰器/工厂函数两种使用形态UDF 类型输入输出抽象基类创建助手标量函数Scalar Function零个、一个或多个标量值一个新的标量值ScalarFunctionudf表函数Table Function零个、一个或多个标量值零个、一个或多个行可结构化类型TableFunctionudtf聚合函数Aggregate Function多行的标量值一个新的标量值AggregateFunctionudaf表聚合函数Table Aggregate Function多行的标量值零个、一个或多个行或结构化类型TableAggregateFunctionudtaf四类函数分别对应 udf.py 中的ScalarFunction第 105 行、TableFunction第 121 行、AggregateFunction第 214 行、TableAggregateFunction第 235 行四个抽象类以及udf第 627 行、udtf第 688 行、udaf第 741 行、udtaf第 789 行四个创建助手。所有用户自定义函数都继承自公共基类UserDefinedFunctionudf.py它定义了三个贯穿函数整个生命周期的默认钩子open(function_context)初始化方法在主工作方法之前被调用适合做一次性设置如建立连接、加载模型。close()收尾方法在最后一次调用主工作方法后被调用用于释放资源。is_deterministic()返回函数结果的确定性。默认返回True只有当调用保证在给定相同参数时永远返回相同结果时才应为真像random()、date()、now()这类非纯函数必须返回False。此外 udf.py 中的FunctionContext类用于在函数内部获取运行时上下文信息提供两个方法get_metric_group()返回当前并行子任务的指标组Metric Group未开启指标时抛出RuntimeError并提示通过python.metric.enabled配置开启。get_job_parameter(key, default_value)按 key 获取全局作业参数的值参数不存在时返回默认值。二、标量函数Scalar Function2.1 语义与抽象基类标量函数将零个、一个或多个标量值映射为一个新的标量值。基类ScalarFunction的核心是一个抽象方法eval(*args)它定义了标量函数的计算逻辑udf.py。2.2 用udf装饰器快速定义推荐使用udf助手函数创建标量函数其完整签名如下udf.pydef udf(fNone, input_typesNone, result_typeNone, deterministicNone, nameNone, func_typegeneral, udf_typeNone)参数说明参数类型说明f可调用对象 /ScalarFunction实例 / 类函数体可为 lambda、普通 Python 函数、functools.partial、实现了__call__的可调用对象或ScalarFunction子类实例input_types可选输入数据类型DataType、strSQL 类型名或其列表也可传RowTyperesult_type必填结果数据类型DataType或strdeterministic可选默认True函数结果的确定性name可选函数名默认取函数__name__或类名func_typegeneral/pandas默认generalPython 函数的执行类型udf_type为 1.12 起废弃的旧参数名会触发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实现有状态逻辑当需要状态或资源时继承ScalarFunction并实现eval、可选实现open/closefrom pyflink.table.udf import ScalarFunction, FunctionContext class SubtractOne(ScalarFunction): def eval(self, i): return i - 1 # 结合 FunctionContext 使用全局作业参数与指标 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 行中验证了 lambda、ScalarFunction子类、可调用对象、functools.partial四种形态的标量函数并通过python.metric.enabled与pipeline.global-job-parameters配置分别验证了指标禁用与全局参数传递两种场景。2.4 确定性、命名与数据类型覆盖test_deterministictest_udf.py验证默认确定性为True当传入deterministicFalse与基类is_deterministic()返回True冲突时构造阶段会抛出ValueError: Inconsistent deterministic。test_name第 744 行验证name参数可覆盖默认函数名。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等全部内置类型的输入输出映射。三、表函数Table Function3.1 语义与抽象基类表函数将零个、一个或多个标量值映射为零行、一行或多行。基类TableFunction的核心同样是抽象方法eval(*args)udf.py但语义上要求以迭代器generator方式产出多行结果。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 侧的PythonTableFunction。3.3 表函数的使用与测试验证表函数在 SQL 中通过LATERAL TABLE(...)使用测试 flink-python/pyflink/table/tests/test_udtf.py 的test_table_function_with_sql_query第 54 行展示了注册与调用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两种执行环境说明表函数在两类模式下均可直接使用。四、聚合函数Aggregate Function4.1 命令式聚合的抽象契约聚合函数将多行的标量值聚合成一个新的标量值。它不再依赖单一的eval而是采用累加器accumulator模式抽象基类为ImperativeAggregateFunctionudf.pyAggregateFunction在其上增加抽象方法get_value(accumulator)第 222-232 行用于物化聚合结果。必须实现的方法方法职责create_accumulator()创建并初始化累加器返回初始值accumulate(accumulator, *args)处理输入值并更新累加器实例get_value(accumulator)每次需要物化聚合结果时调用返回值可以是早到的、不完整的结果数据到达时周期性下发也可以是最终聚合结果可选方法retract(accumulator, *args)从累加器中撤回之前累计的输入值用于处理流式场景中的更新与回撤如GROUP BY上的流聚合。默认抛出RuntimeError(Method retract is not implemented)。merge(accumulator, accumulators)将一组累加器合并为一个。对于无界会话窗口分组聚合和有界分组聚合必须实现注意accumulator中可能已含有之前的聚合结果不能替换或清空该实例。get_result_type()/get_accumulator_type()返回结果与累加器的DataType实现类通常覆写它们或在使用udaf时通过参数显式指定。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 提供了多个可直接参考的完整实现例如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())要点累加器类型与结果类型既可以覆写get_accumulator_type()/get_result_type()提供也可以在udaf参数中显式传入两者需同时为DataType或同时为str否则抛出TypeErrorudf.py。func_typepandas时udaf将 Python 函数包装为 Pandas UDAF此时结果类型不能是MapType源码中显式抛出TypeError且累加器会被自动构造为ARRAYresult_type。udaf(result_typeDataTypes.FLOAT(), func_typepandas) def mean_udaf(v): return v.mean()4.3 测试覆盖的进阶特性test_udaf.py 验证了大量实战场景test_double_aggregate第 254 行同一查询中多个 UDAF 组合使用。test_mixed_with_built_in_functions_with_retract/without_retract第 277/326 行UDAF 与内置函数混合以及是否实现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 行状态清理与 DataView 的clear()语义。五、表聚合函数Table Aggregate Function5.1 语义与抽象方法表聚合函数将多行的标量值映射为零行、一行或多行或结构化类型。如果输出记录只包含一个字段可以省略结构化记录直接发射标量值运行时会自动将其包装进一行。基类TableAggregateFunction在ImperativeAggregateFunction之上定义抽象方法emit_value(accumulator)udf.py它返回一个可迭代对象迭代器/生成器在每次需要物化聚合结果时被调用。5.2 用udtaf装饰器定义udtaf的完整签名udf.py与udaf一致但仅支持func_typegeneral传入pandas会抛出ValueError。官方示例udf.py——一个经典的 Top2 实现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中同时实现了retract与merge使其可用于流式回撤聚合与有界/会话窗口聚合场景。emit_value每次产出多行这正是表聚合与普通聚合的核心差异。六、DataView大数据量累加器的性能利器6.1 为什么需要 DataView官方文档udf.rst 的 DataView 一节明确说明如果累加器需要存储大量数据应当使用pyflink.table.ListView和pyflink.table.MapView替代 list 和 dict。这两种数据结构提供与 list/dict 类似的功能但通常具有更好的性能——它们借助 Flink 的状态后端state backend避免了对每个元素都产生不必要的状态访问开销。6.2 ListView 与 MapView 的 API定义在 flink-python/pyflink/table/data_view.pyListView第 43-102 行提供get()、add(value)、add_all(values)、clear()支持迭代__iter__可视为状态后端的 list。MapView第 105-197 行提供get(key)、put(key, value)、put_all(dict)、remove(key)、contains(key)、items()、keys()、values()、is_empty()、clear()并支持[]下标读写__getitem__/__setitem__/__delitem__与in判断__contains__可视为状态后端的 dict。两者均继承DataView抽象类并实现clear()data_view.py。6.3 在累加器中声明并使用在累加器类型中通过DataTypes.LIST_VIEW(...)与DataTypes.MAP_VIEW(...)声明。这两个类型工厂定义于 flink-python/pyflink/table/types.py官方文档强调它们只能用于聚合函数的累加器类型声明。以测试 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实现高效的去重计数累加器类型声明为ROWf0 MAPSTRING, STRING, f1 BIGINT并演示了is_empty()、下标读写、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相关测试覆盖test_list_view第 369 行、test_map_view第 391 行、test_data_view_clear第 425 行验证聚合状态下调clear()后的行为、test_map_view_iterate第 442 行验证items()/keys()/values()迭代。七、源码级运行机制7.1 从 Python 函数到 Java UDF 的桥接所有udf/udtf/udaf/udtaf返回的对象都是UserDefinedFunctionWrapper的子类包装器udf.py。其内部机制输入类型归一化input_types支持RowType、单个DataType/str或列表统一转为列表非法类型抛出TypeError第 336-349 行。确定性检查若显式传入deterministic且与函数自身is_deterministic()冲突抛出ValueError第 356-359 行未显式指定时默认取函数自身判定纯 Python 函数默认True第 361-363 行。函数调用包装器实现__call__内部通过expr.call(self, *args)将函数调用转化为 Table API 的Expression第 368-370 行因此 UDF 可以直接参与 Table 表达式运算与 SQL 解析。alias()与行输入模式alias(*names)为表函数结果列命名第 372-374 行_set_takes_row_as_input()标记以整行作为输入的模式。序列化与 Java 桥接_java_user_defined_function()第 380-411 行通过cloudpickle.dumps()序列化 Python 函数并根据func_type映射到 Java 侧PythonFunctionKind.GENERAL或PANDAS最终构造 Java 侧的PythonScalarFunction/PythonTableFunction/PythonAggregateFunction/PythonTableAggregateFunction对应org.apache.flink.table.functions.python包通过PythonEnv.ExecType.PROCESS指定以独立进程方式执行 Python 代码第 601-605 行。7.2 lambda 与普通函数的委托包装当f不是UserDefinedFunction实例如 lambda、普通函数时包装器通过_create_delegate_function()生成内部委托实现DelegatingScalarFunction第 258-268 行、DelegationTableFunction第 271-281 行、DelegatingPandasAggregateFunction第 284-300 行get_value直接取累加器首个元素从而使装饰器语法与类继承语法在底层统一。八、函数注册与使用自定义函数创建后可通过以下方式注册到 Table 环境from pyflink.table import EnvironmentSettings, TableEnvironment env_settings EnvironmentSettings.in_streaming_mode() t_env TableEnvironment.create(env_settings) # 注册为临时系统函数SQL 可直接调用 t_env.create_temporary_system_function(add_one, add_one) # 或注册为临时目录函数 t_env.create_temporary_function(add_one, add_one) # SQL 中使用 t_env.sql_query(SELECT add_one(a) FROM MyTable) # Table API 表达式中使用 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 行演示了注册、调用与drop_function删除的完整生命周期test_overwrite_builtin_function第 233 行还展示了通过注册同名 UDF 覆盖内置函数如重定义plus的能力。九、最佳实践与常见陷阱必须指定result_type或覆写get_result_typeudf的result_type是必填参数类型既可用DataTypes.xxx()也可用 SQL 类型字符串如BIGINT字符串形式在 udf.py 中被直接透传给 Java 侧解析。确定性声明影响优化非纯函数务必设置deterministicFalse如随机数、取当前时间否则 Flink 可能基于错误假设做查询优化同时注意deterministic与is_deterministic()冲突会直接报错。流式聚合实现retract默认retract抛异常涉及回撤如GROUP BY流聚合、与内置聚合函数混合时需实现参考 test_udaf.py 的混合场景测试。会话窗口/有界聚合实现merge无界会话窗口分组聚合与有界分组聚合要求实现merge且不能替换传入的accumulator实例。大数据量累加器优先用ListView/MapView按官方文档建议替代裸 list/dict 以获得状态后端优化DataTypes.LIST_VIEW/MAP_VIEW只能用于聚合函数累加器类型声明。资源生命周期重资源连接、模型在open()中初始化、close()中释放全局参数通过FunctionContext.get_job_parameter()读取指标需先开启python.metric.enabled。十、参考文档与源码索引官方参考文档flink-python/docs/reference/pyflink.table/udf.rst核心实现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赞分享大数据流处理批处理数据工程【免费下载链接】flink项目地址https://gitcode.com/gh_mirrors/fli/flink点击查看免费下载相关推荐Flink Python Table API 用户自定义函数完全指南UDF、UDTF、UDAGG 与 UDTAGG 实战Flink Python Table API 用户自定义函数完全指南UDF、UDTF、UDAGG 与 UDTAGG 实战 本指南围绕 Flink PyFlin大数据流处理批处理数据工程Flink自定义函数开发UDF、UDAF、UDTF编写与注册完整指南Flink自定义函数开发UDF、UDAF、UDTF编写与注册完整指南 Apache Flink作为业界领先的流处理框架提供了强大的自定义函数功能允许开发者示例工程大数据PyFlink 用户自定义函数UDF / UDTF / UDAGG / UDTAGG完整实战指南从 Python Table API 到 SQL 的四种函数定义与调用PyFlink 用户自定义函数UDF / UDTF / UDAGG / UDTAGG完整实战指南从 Python Table API 到 SQL 的四种函大数据流处理批处理数据工程上一篇精通Google API Python Client从基础认证到高级批量请求处理完整指南下一篇Spray高级技巧绕过账户锁定策略的10种方法创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考