。

本节我们将介绍最后一种函数#xff1a;UDTAF——用户自定义表值聚合函数。

UDTAF

UDTAFTableAggregateFunction的实现累加器定义创建累加

返回类型计算

在前面几篇文章中我们分别介绍了UDF、UDTF和UDAF这三种用户自定义函数。

本节我们将介绍最后一种函数UDTAF——用户自定义表值聚合函数。

UDTAF

UDTAF函数即具备了UDTF的特点也具备UDAF的特点。

即它可以像《0基础学习PyFlink——用户自定义函数之UDTF》介绍的UDTF那样可以返回任意数量的行作为输又可以像《0基础学习PyFlink——用户自定义函数之UDAF》介绍的UDAF那样通过聚合的数据多组计算出一个值。

举一个例子我们拿到一个学生成绩表每行包括

将一行中的“英语成绩”和“数学成绩”拆成“成绩”和“科目”相当于把一行数据拆解成多行如上图左侧“张三”只有一行而右侧有两行“张三”信息。

这种拆解操作就需要T类型的用户自定义函数比如UDTF和UDTAF。

而我们需要计算一个年级一科的平均成绩比如1年级英语的平均成绩则需要按年级聚合之后再做计算。

这个就需要A类型的用户自定义函数比如UDAF和UDTAF。

同时要满足上述两种技术方案的就是UDTAF。

我们先看下主体代码它和《0基础学习PyFlink——用户自定义函数之UDAF》中的很像。

但是有两个重要区别要设置成in_streaming_mode模式否则会报错udtaf要修饰一个对象而非一个方法

def

fileconfig.set_string(parallelism.default,

1)env_settings

TableEnvironment.create(env_settings)row_type_tab_source

DataTypes.ROW([DataTypes.FIELD(name,

DataTypes.STRING()),

DataTypes.STRING())])students_score

[(张三,

t_env.from_elements(students_score,

row_type_tab_source)split_class

udtaf(SplitClass())tab_source.group_by(col(grade))

\.select(col(*))

\.execute().print()TableAggregateFunction的实现

用于计算的类要继承于TableAggregateFunction即UDTAF中的TAF。

class

SplitClass(TableAggregateFunction):_class_keys

[english,

math]我们需要通过get_result_type告诉框架UDTAF函数返回的是什么类型的数据。

一般我们都是构造一个行类型——ROW然后定义其每个字段的值和类型

namestring类型用户姓名scorefloat类型考分avg

scorefloat类型科目年级平均分数classsting类型科目名称

累加器

accumulator累加器是用于参与计算的中间数据。

比如这个案例中我们会向让accumulator保存拆解后的数据即一行拆解成多行后的数据然后再计算各年级每科的平均成绩。

def

get_accumulator_type(self):return

DataTypes.ARRAY(DataTypes.ROW([DataTypes.FIELD(name,

DataTypes.STRING()),

namestring类型姓名scorefloat类型分数classstring类型科目名称

def

create_accumulator(self):return

[]累加

self._class_keys:accumulator.append(Row(row[name],

row[i],

DataTypes.ROW([DataTypes.FIELD(name,

DataTypes.STRING()),

DataTypes.STRING())])可以看到result_type返回类型和accumulator_type累加器类型是不一样的也可以一样主要看怎么计算规则。

前者比后者多了“学科年级平均分”avg

这些字段和我们目标字段只差一个grade年级。

因为原始表中有grade且我们会通过grade聚类所以最终我们可以获得这个信息而不用在这儿定义。

需要注意的是虽然表值类型函数返回的是一组数据若干Row但是这儿只是返回Row的具体定义而不是ARRAY[Row]。

def

x这个函数会在最后执行它会通过累加器中的数据计算“学科年级平均分”然后构造和“返回类型”一直的Row到rows数组中。

最后通过yeild关键字返回一个生成器我们可以将其看成还是一组Row即拆解后的结果。

最后我们看下结果

--------------------------------------------------------------------------------------------------------------------------------------------------------------------

grade

--------------------------------------------------------------------------------------------------------------------------------------------------------------------

80.0

--------------------------------------------------------------------------------------------------------------------------------------------------------------------

rows

udf,udtf,udaf,udtaf,TableAggregateFunction

import

SplitClass(TableAggregateFunction):_class_keys

[english,

create_accumulator(self):return

[]def

self._class_keys:accumulator.append(Row(row[name],

row[i],

get_accumulator_type(self):return

DataTypes.ARRAY(DataTypes.ROW([DataTypes.FIELD(name,

DataTypes.STRING()),

DataTypes.ROW([DataTypes.FIELD(name,

DataTypes.STRING()),

fileconfig.set_string(parallelism.default,

1)env_settings

TableEnvironment.create(env_settings)row_type_tab_source

DataTypes.ROW([DataTypes.FIELD(name,

DataTypes.STRING()),

DataTypes.STRING())])students_score

[(张三,

t_env.from_elements(students_score,

row_type_tab_source)split_class

udtaf(SplitClass())tab_source.group_by(col(grade))

\.select(col(*))