Flink SQL 表值聚合函数(Table Aggregate Function)详解

使用场景: 表值聚合函数即 UDTAF,这个函数⽬前只能在 Table API 中使⽤,不能在 SQL API 中使⽤。

函数功能:

在 SQL 表达式中,如果想对数据先分组再进⾏聚合取值:

select max(xxx) from source_table group by key1, key2

上⾯ SQL 的 max 语义产出只有⼀条最终结果,如果想取聚合结果最⼤的 n 条数据,并且 n 条数据,每⼀条都要输出⼀次结果数据,上⾯的 SQL 就没有办法实现了。

所以 UDTAF 为了处理这种场景,可以⾃定义 怎么取 , 取多少条 最终的聚合结果,UDTAF 和 UDAF 是类似的。

在这里插入图片描述

案例场景: 有⼀个饮料表有 3 列,分别是 id、name 和 price,⼀共有 5 ⾏,需要找到价格最⾼的两个饮料,类似于 top2,表值聚合函数,需要遍历所有 5 ⾏数据,输出结果为 2 ⾏数据的⼀个表。

开发流程:

实现 TableAggregateFunction 接⼝,其中所有的⽅法必须是 public 的、⾮ static 的

必须实现以下⽅法:

Acc聚合中间结果 createAccumulator() : 为当前 Key 初始化⼀个空的 accumulator,存储了聚合的中间结果,⽐如在执⾏ max() 时会存储每⼀条中间结果的 max 值;

accumulate(Acc accumulator, Input输⼊参数) : 每⼀⾏数据,都会调⽤ accumulate() ⽅法更新 accumulator,⽅法对每⼀条输⼊数据执⾏,⽐如执⾏ max() 时,遍历每⼀条数据执⾏;这个⽅法必须声明为 public 和⾮ static 的,accumulate ⽅法可以重载,每个⽅法的参数类型可以不同,⽀持变⻓参数。

emitValue(Acc accumulator, Collector collector) 或者 emitUpdateWithRetract(Acc accumulator, RetractableCollector collector) :

当所有的数据处理完之后,调⽤ emit ⽅法来计算和输出最终结果,可以⾃定义输出多少条以及怎样输出结果。

对于 emitValue 以及 emitUpdateWithRetract 区别,以 TopN 举例,emitValue 每次都会发送所有的最⼤的 n 个值,⽽这在流式任务中会有性能问题,为提升性能,可以实现 emitUpdateWithRetract ⽅法,这个⽅法在 retract 模式下会增量输出结果,⽐如只在有数据更新时,做到撤回⽼数据,再发送新数据,⽽不需要每次都发出全量的最新数据。

如果同时定义了 emitUpdateWithRetract、emitValue ⽅法,那 emitUpdateWithRetract 会优先于 emitValue ⽅法被使⽤,因为引擎会认为 emitUpdateWithRetract 会更加⾼效,它的输出是增量的。

某些场景下必须实现:

  • retract(Acc accumulator, Input输⼊参数) : 回撤流的场景必须实现,在计算回撤数据时调⽤,如果没有实现则会直接报错。
  • merge(Acc accumulator, Iterable it) : 在批式聚合以及流式聚合中的 Session、Hop 窗⼝聚合场景必须实现,这个⽅法对优化也有帮助,例如,打开了两阶段聚合优化,需要 AggregateFunction 实现 merge ⽅法,从⽽在第⼀阶段先进⾏数据聚合。
  • resetAccumulator() : 在批式聚合中是必须实现的。

关于⼊参、出参数据类型:

默认情况下,⽤户的 Input输⼊参数( accumulate(Acc accumulator, Input输⼊参数) 的⼊参 Input输⼊参数 )、accumulator( Acc聚 合中间结果 createAccumulator() 的返回结果)、 Output输出参数 数据类型( emitValue(Acc acc,Collector<Output输出参数> out) 的 Output输出参数 )会被 Flink 反射获取,但对于accumulator 和 Output输出参数类型来说,Flink SQL 的类型推导在遇到复杂类型的时候可能会推导出错误的结果(注意: Input输⼊参数 因为是上游算⼦传⼊的,所以类型信息是确认的,不会出现推导错误的情况),⽐如那些⾮基本类型 POJO 的复杂类型,所以跟 ScalarFunction 和 TableFunction ⼀样, AggregateFunction 提供了TableAggregateFunction#getResultType() 和 TableAggregateFunction#getAccumulatorType() 来分别指定最终返回值类型和accumulator 的类型,两个函数的返回值类型都是 TypeInformation。

  • getResultType() : 即 emitValue(Acc acc, Collector<Output输出参数> out) 的输出结果数据类型;
  • getAccumulatorType() : 即 Acc聚合中间结果 createAccumulator() 的返回结果数据类型;

案例场景: Top2

定义⼀个 TableAggregateFunction 来计算给定列的最⼤的 2 个值

在 TableEnvironment 中注册函数

在 Table API 查询中使⽤函数(当前只在 Table API 中⽀持 TableAggregateFunction)

实现思路:

计算最⼤的 2 个值,accumulator 需要保存当前的最⼤的 2 个值,定义了类 Top2Accum 作为 accumulator,Flink 的 checkpoint 机制会⾃动保存 accumulator,在失败时进⾏恢复,来保证精确⼀次的语义。

Top2 表值聚合函数(TableAggregateFunction)的 accumulate() ⽅法有两个输⼊,第⼀个是 Top2Accum accumulator,另⼀个是⽤户定义的输⼊:输⼊的值 v,尽管 merge() ⽅法在⼤多数聚合类型中不是必须的,但在样例中提供了它的实现。并且定义了 getResultType() 和 getAccumulatorType() ⽅法。

代码案例:

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.util.Collector;/*** 输入数据:* a,1* a,2* a,3* * 输出结果:* res=>:1> +I[a, 1, 1]* res=>:1> -D[a, 1, 1]* res=>:1> +I[a, 2, 1]* res=>:1> +I[a, 1, 2]* res=>:1> -D[a, 2, 1]* res=>:1> -D[a, 1, 2]* res=>:1> +I[a, 3, 1]* res=>:1> +I[a, 2, 2]*/
public class TableAggregateFunctionTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);DataStreamSource<String> source = env.socketTextStream("localhost", 8888);SingleOutputStreamOperator<Tuple2<String,Integer>> tpStream = source.map(new MapFunction<String, Tuple2<String,Integer>>() {@Overridepublic Tuple2<String,Integer> map(String input) throws Exception {return new Tuple2<>(input.split(",")[0],Integer.parseInt(input.split(",")[1]));}});tEnv.registerFunction("top2", new Top2());Table table = tEnv.fromDataStream(tpStream, "key,value");tEnv.createTemporaryView("SourceTable", table);// 使⽤函数Table res = tEnv.from("SourceTable").groupBy("key").flatAggregate("top2(value) as (v, rank)").select("key, v, rank");tEnv.toChangelogStream(res).print("res=>");env.execute();}/*** Accumulator for Top2.*/public static class Top2Accum {public Integer first;public Integer second;}public static class Top2 extends TableAggregateFunction<Tuple2<Integer, Integer>, Top2Accum> {@Overridepublic Top2Accum createAccumulator() {Top2Accum acc = new Top2Accum();acc.first = Integer.MIN_VALUE;acc.second = Integer.MIN_VALUE;return acc;}public void accumulate(Top2Accum acc, Integer v) {if (v > acc.first) {acc.second = acc.first;acc.first = v;} else if (v > acc.second) {acc.second = v;}}public void merge(Top2Accum acc, java.lang.Iterable<Top2Accum> iterable) {for (Top2Accum otherAcc : iterable) {accumulate(acc, otherAcc.first);accumulate(acc, otherAcc.second);}}public void emitValue(Top2Accum acc, Collector<Tuple2<Integer, Integer>> out) {// emit the value and rankif (acc.first != Integer.MIN_VALUE) {out.collect(Tuple2.of(acc.first, 1));}if (acc.second != Integer.MIN_VALUE) {out.collect(Tuple2.of(acc.second, 2));}}}
}

测试结果:

在这里插入图片描述

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/140801.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

2022年06月 Python(五级)真题解析#中国电子学会#全国青少年软件编程等级考试

Python等级考试(1~6级)全部真题・点这里 一、单选题(共25题,每题2分,共50分) 第1题 Python中 print(“八进制{: o}”.format(12)) 正确的输出结果是?( ) A: 八进制:O B: 八进制:O14 C: 八进制14O D: 八进制14 答案:D 字符串的format()格式。 第2题 下列的程…

PostgreSQL基础入门

为什么选择PostgreSQL 功能更全面&#xff1a;PGSQL的功能更加全面&#xff0c;支持开窗函数、物化视图、分区表、json等类型&#xff0c;MySQL8以上支持开窗函数、分区表、json等&#xff0c;但物化视图仍不支持。高可用&#xff1a;PG更适合分布式环境&#xff0c;如流复制、…

pychon/PIL/opencv/json学习过程中遇到的问题

1. 使用PIL.Image读取图片 注意&#xff1a;pytorch中对图像预处理是transforms的输入必须是PIL格式的文件&#xff0c;使用cv2读取的图片就按照第二条的代码处理&#xff08;3通道合并、归一化处理&#xff09; from PIL import Image img Image.open("test1.jpg"…

TensorFlow: An open-source software library for Machine Intelligence

Google开源的机器学习软件包 项目主页:https://www.tensorflow.org/Github主页:GitHub - tensorflow/tensorflow: An Open Source Machine Learning Framework for Everyone主要版本:1.0(当前最新版本),0.12.0开源协议:Apache 2.0TensorFlow™ 是一个采用数据流图(data…

2023 年最新企业微信官方会话机器人开发详细教程(更新中)

目标是开发一个简易机器人&#xff0c;能接收消息并作出回复。 获取企业 ID 企业信息页面链接地址&#xff1a;https://work.weixin.qq.com/wework_admin/frame#profile 自建企业微信机器人 配置机器人应用详情 功能配置 接收消息服务器配置 配置消息服务器配置 配置环境变量…

如何利用 cpolar 内网穿透技术实现 U8 用友 ERP 异地访问

文章目录 前言1. 服务器本机安装U8并调试设置2. 用友U8借助cpolar实现企业远程办公2.1 在被控端电脑上&#xff0c;点击开始菜单栏&#xff0c;打开设置——系统2.2 找到远程桌面2.3 启用远程桌面 3. 安装cpolar内网穿透3.1 注册cpolar账号3.2 下载cpolar客户端 4. 获取远程桌面…

【MATLAB源码-第75期】基于模拟退火算法(SA)的栅格路径规划,输出做短路径图和适应度曲线。

操作环境&#xff1a; MATLAB 2022a 1、算法描述 模拟退火算法是一种启发式优化算法&#xff0c;通常用于解决组合优化问题&#xff0c;例如旅行商问题和图着色问题。它模拟了固体材料在退火过程中逐渐冷却达到稳定状态的行为&#xff0c;以寻找问题的全局最优解。 以下是模…

Hadoop原理,HDFS架构,MapReduce原理

Hadoop原理&#xff0c;HDFS架构&#xff0c;MapReduce原理 2022找工作是学历、能力和运气的超强结合体&#xff0c;遇到寒冬&#xff0c;大厂不招人&#xff0c;可能很多算法学生都得去找开发&#xff0c;测开 测开的话&#xff0c;你就得学数据库&#xff0c;sql&#xff0c…

postswigger 靶场(CSRF)攻略-- 1.没有防御措施的 CSRF 漏洞

靶场地址&#xff1a; What is CSRF (Cross-site request forgery)? Tutorial & Examples | Web Security Academy (portswigger.net)https://portswigger.net/web-security/csrf 没有防御措施的 CSRF 漏洞 题目中已告知易受攻击的是电子邮件的更改功能&#xff0c;而目…

Oracle Primavera Unifier 23.10 新特征

根据官方的说法&#xff0c;Unifier 23.7 ~ 23.9 更多为对功能bug的修复&#xff0c;以下将对23.10进行重点介绍 Cost Sheets Cost Sheets Support Conditional Formatting Conditional formatting of table data is now supported in cost sheets with features such as ce…

基于注解的声明式事务

1.什么是事务 数据库事务(transaction)是访问并可能操作各种数据项的一个数据库操作序列&#xff0c;这些操作要么全部执行要么全部不执行&#xff0c;是一个不可分割的工作单位。事务由事务开始与事务结束之间执行的全部数据库操作组成。 2.事务的特性 A&#xff1a;原子性(A…

java Stream编程笔记

文章目录 Stream介绍什么是 Stream&#xff1f; Stream中间操作过滤操作&#xff08;filter&#xff09;映射操作&#xff08;map&#xff09;排序操作&#xff08;sorted&#xff09;截断操作&#xff08;limit 和 skip&#xff09; Stream 的终止操作forEach 和 peek聚合操作…

【pytorch深度学习】torch-张量Tensor

torch-张量Tensor 文章目录 torch-张量Tensor1. 张量Tensor 1. 张量Tensor torch.tensor() # 创建一个标量&#xff08;0维张量&#xff09; scalar_tensor torch.tensor(3.14) # 创建一个向量&#xff08;1维张量&#xff09; vector_tensor torch.tensor([1, 2, 3]) # 创…

【FPGA】十进制计数器 | 实现 4-bit 2421 十进制计数器 | 有限状态机(FSM)

目录 Ⅰ. 实践说明 0x00 十进制计数器 0x01 有限状态机&#xff08;FSM&#xff09; Ⅱ. 实践部分 0x00 4-bit 2421 十进制计数器 Ⅰ. 实践说明 0x00 十进制计数器 十进制计数器是一种以十进制运算的计数器&#xff0c;从 0 数到 9&#xff0c;然后返回 0 状态。由于它需…

青少年编程学习 等级考试 蓝桥杯/NOC/GESP等比赛资料合集

一、博主愚见 在当今信息技术高速发展的时代&#xff0c;编程已经成为了一种必备的技能。随着社会对于科技人才的需求不断增加&#xff0c;青少年编程学习正逐渐成为一种趋势。为了更好地帮助青少年学习编程&#xff0c;提升他们的技能和素质&#xff0c;博主结合自身多年从事青…

如何使用CORS和CSP保护前端应用程序安全

前端应用在提供无缝用户体验方面起着核心作用。在当今互联网的环境中&#xff0c;第三方集成和API的普及使得确保强大的安全性至关重要。安全漏洞可能导致数据盗窃、未经授权访问以及品牌声誉受损。本文将向您展示如何使用CORS和CSP为您的网页增加安全性。 嗨&#xff0c;大家好…

大数据可视化数据大屏可视化模板【可视化项目案例-05】

🎉🎊🎉 你的技术旅程将在这里启航! 🚀🚀 本文选自专栏:可视化技术专栏100例 可视化技术专栏100例,包括但不限于大屏可视化、图表可视化等等。订阅专栏用户在文章底部可下载对应案例源码以供大家深入的学习研究。 🎓 每一个案例都会提供完整代码和详细的讲解,不…

半导体PLM解决方案助力企业提升竞争力 | 百世慧®

行业背景 半导体&#xff08;semiconductor&#xff09;指常温下导电性能介于导体与绝缘体之间的材料。 半导体产业主要涵盖集成电路、处理器、存储器、传感器及光电子设备等领域。产品则广泛应用在个人电脑、智能手机、平板电脑、汽车电子、医疗设备、工业控制等领域。 近些…

基于安卓android微信小程序的校园互助平台

项目介绍 随着社会的发展&#xff0c;社会的方方面面都在利用信息化时代的优势。互联网的优势和普及使得各种系统的开发成为必需。 本文以实际运用为开发背景&#xff0c;运用软件工程原理和开发方法&#xff0c;它主要是采用java语言技术和mysql数据库来完成对系统的设计。整…

Unity中Shader的间接光的产生Meta Pass

文章目录 前言Unity中Shader的间接光的产生Meta Pass&#xff0c;这也是属于全局光照 GI 的内容。主要实现像现实生活中&#xff0c;光线照到有颜色的物体后&#xff0c;该物体有反射出该颜色的光的效果。 一、我们先使用Unity自带的Shader看看间接光效果1、先按照如下设置搭建…