Ctrl+K
Logo image Logo image

Site Navigation

  • API Reference
  • Examples

Site Navigation

  • API Reference
  • Examples

Section Navigation

  • PyFlink Table
    • TableEnvironment
    • Table
    • Data Types
    • Window
    • Expressions
    • User Defined Functions
    • Descriptors
    • StatementSet
    • Catalog
  • PyFlink DataStream
  • PyFlink Common

pyflink.table.udf.udtaf#

udtaf(f: Optional[Union[Callable, pyflink.table.udf.TableAggregateFunction, Type]] = None, input_types: Optional[Union[List[pyflink.table.types.DataType], pyflink.table.types.DataType, str, List[str]]] = None, result_type: Optional[Union[pyflink.table.types.DataType, str]] = None, accumulator_type: Optional[Union[pyflink.table.types.DataType, str]] = None, deterministic: Optional[bool] = None, name: Optional[str] = None, func_type: str = 'general') → Union[pyflink.table.udf.UserDefinedAggregateFunctionWrapper, Callable][source]#

Helper method for creating a user-defined table aggregate function.

Example:

>>> # The input_types is optional.
>>> 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 'ARRAY<BIGINT>'
...
...     def get_result_type(self):
...         return 'ROW<a BIGINT>'
>>> top2 = udtaf(Top2())
Parameters
  • f – user-defined table aggregate function.

  • input_types – optional, the input data types.

  • result_type – the result data type.

  • accumulator_type – optional, the accumulator data type.

  • deterministic – the determinism of the function’s results. True if and only if a call to this function is guaranteed to always return the same result given the same parameters. (default True)

  • name – the function name.

  • func_type – the type of the python function, available value: general (default: general)

Returns

UserDefinedAggregateFunctionWrapper or function.

New in version 1.13.0.

previous

pyflink.table.udf.TableAggregateFunction

next

pyflink.table.data_view.ListView

Show Source

Created using Sphinx 4.5.0.