关注并标星 3分钟秒懂大数据
每天1次，打卡阅读
更快更全面获取大数据技术博客
UDAF介绍与使用
自定义函数类别
UDF: 自定义标量函数(User Defined Scalar Function)。一行输入一行输出。
UDAF: 自定义聚合函数。多行输入一行输出。
UDTF: 自定义表函数。一行输入多行输出或一列输入多列输出。
UDAF，自定义聚合函数，需要继承AggregateFunction<T,ACC>抽象类，实现一系列方法。AggregateFunction抽象类如下：
abstract class AggregateFunction<T, ACC>extends UserDefinedAggregateFunction<T, ACC>T: 表示UDAF最终输出的结果类型ACC: 表示UDAF存放中间结果的类型
最基本的UDAF至少需要实现如下三个方法:
需要先定义一个Accumulator类，类里面定义变量，存放聚合的中间结果
createAccumulator: createAccumulator 方法是用来初始化你定义的Accumulator类，将内部定义的变量赋值为空或者0。
accumulate: 定义如何根据输入更新Accumulator，主要是编写中间的逻辑代码，根据输入变量来更新你的输出中间变量。
getValue: 定义如何返回Accumulator中存储的中间结果作为UDAF的最终结果。
UDAF使用案例
source源：Json格式的数据
客户需求： 需要根据productPrice 的数据进行累加，输入如上图数据，1、2、3、4、5 输出为 1、3、6、10、15,这个就相当于多行输入，最后只输出一行数据。
（1）编写Flink UDAF代码
pom.xml 如下：
<groupId>com.flink-ray</groupId><artifactId>udaf</artifactId><version>1.0-SNAPSHOT</version><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><maven.compiler.source>1.8</maven.compiler.source><maven.compiler.target>1.8</maven.compiler.target><scala.binary.version>2.11</scala.binary.version><flink.version>1.12.1</flink.version></properties><dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-container_${scala.binary.version}</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-runtime_${scala.binary.version}</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_${scala.binary.version}</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_${scala.binary.version}</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-common</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner_${scala.binary.version}</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-runtime-blink_${scala.binary.version}</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-python_${scala.binary.version}</artifactId><version>${flink.version}</version><scope>provided</scope></dependency></dependencies><build><plugins><!-- Scala Compiler --><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.2.2</version><executions><execution><id>scala-compile-first</id><phase>process-resources</phase><goals><goal>compile</goal></goals></execution><execution><id>scala-test-compile</id><phase>process-test-resources</phase><goals><goal>testCompile</goal></goals></execution></executions><configuration><args><arg>-nobootcp</arg></args></configuration></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-shade-plugin</artifactId><version>3.0.0</version><executions><!-- Run shade goal on package phase --><execution><phase>package</phase><goals><goal>shade</goal></goals><configuration><artifactSet><excludes><exclude>org.apache.flink:force-shading</exclude><exclude>com.google.code.findbugs:jsr305</exclude><!-- <exclude>org.slf4j:*</exclude>--><!-- <exclude>log4j:*</exclude>--></excludes></artifactSet><filters><filter><!-- Do not copy the signatures in the META-INF folder.Otherwise, this might cause SecurityExceptions when using the JAR. --><artifact>*:*</artifact><excludes><exclude>META-INF/*.SF</exclude><exclude>META-INF/*.DSA</exclude><exclude>META-INF/*.RSA</exclude></excludes></filter></filters><transformers><transformerimplementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"><mainClass></mainClass></transformer></transformers></configuration></execution></executions></plugin></plugins></build>
（2）编写UDAF时，一定要定义数据的输入输出类型，通过FunctionHint来接收
package com.flink_ray;import org.apache.flink.table.annotation.DataTypeHint;import org.apache.flink.table.annotation.FunctionHint;import org.apache.flink.table.functions.AggregateFunction;//自定义UDAF(input = ("INT"), output = ("INT"))public class UDAFSum extends AggregateFunction<Integer, UDAFSum.SumAccumulator> {//定义一个Accumulator，存放聚合的中间结果public static class SumAccumulator{public int sumPrice;}/** 初始化Accumulator* @return*/public SumAccumulator createAccumulator() {SumAccumulator sumAccumulator = new SumAccumulator();sumAccumulator.sumPrice=0;return sumAccumulator;}/* 定义如何根据输入更新Accumulator* @param accumulator Accumulator* @param productPrice 输入*/public void accumulate( SumAccumulator accumulator, int productPrice){accumulator.sumPrice += productPrice;}// * 返回聚合的最终结果// * @param accumulator Accumulator// * @returnpublic Integer getValue(SumAccumulator accumulator) {return accumulator.sumPrice;}}
（3）打成Jar包
编写完后，打成jar包
mvn package -Dcheckstyle.skip以上就是Flink UDAF的使用教程！觉得好的，点赞，在看，分享三连击，谢谢！！！
End
找各类大数据技术文章和面经，就来
<3分钟秒懂大数据>
随时更新互联网大数据组件内容
专为学习者提供技术博文
快和身边的小伙伴一起关注我们吧！
作者简介：逆流而上Mr李，秋招7offer,CSDN博客：https://blog.csdn.net/weixin_38201936
原文作者：3分钟秒懂大数据；如有侵权，请联系管理员删除。
请登录后发表评论
登录
全部评论
最新文章
- 大数据面试，被问到频率最高的三道考题，值得一看~
- 工作中常用的50+ Linux 命令，建议收藏~
- 字节面试sql题：有一张工资表，字段分别是：员工工号、部门、工资。要求各部门除去最高工资和最低工资后的平均工资
- 一篇忍不住看很多遍的数据建模理论与实践(7000字+25张图)
- 网易云音乐P0级事故！官方回应：未删库，未跑路！
- 我发现很多计算机专业的学生都长期处于迷茫中...
- 离职流程都走完了，领导又分了一个新任务让我要么工作日加班，要么周末加班，两周搞完...
- HR最认可的5个离职原因：1、个人发展空间受限；2、公司组织结构变化；3、工作内容与职业发展不匹配...
- 月薪4.5w，离职后每月只补偿2590元，违约就要赔100万！
- 这是雷军的简历，落魄时卡里只有冰冷的40亿...
