【Spark精讲】一文讲透SparkSQL聚合过程以及UDAF开发

2023-12-31 07:04

本文主要是介绍【Spark精讲】一文讲透SparkSQL聚合过程以及UDAF开发,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

SparkSQL聚合过程

这里的 Partial 方式表示聚合函数的模式,能够支持预先局部聚合,这方面的内容会在下一节详细介绍。 对应实例中的聚合语句,因为 count 函数支持 Partial 方式,因此调用的是 planAggregateWithoutDistinct 方法,生成了图 7.4 中的两个 HashAggregate (聚合执行方式中的一种,后续详细介绍)物理算子树节点,分别进行局部聚合与最终的聚合。 最后,在生成的 SparkPlan 中添加 Exchange 节点,统一排序与分区信息,生成物理执行计划(ExecutedPlan)。

聚合查询在计算聚合值的过程中,通常都需要保存相关的中间计算结果,例如 max 函数需要保存当前最大值, count 函数需要保存当前的数据总数,求平均值的 avg 函数需要同时保存 count 和 sum 的值,更复杂的函数(如 pencent让等)甚至需要临时存储全部的数据 。 聚合查询 计算过程中产生的这些中间结果会临时保存在聚合函数缓冲区。

在 SparkSQL 中,聚合过程有 4种模式,分别是 Partial模式、 ParitialMerge模式、 Final模式 和 Complete模式。

Final模式一般和 Partial模式组合在一起使用。 Partial模式可以看作是局部数据的聚合,在 具体实现中, Partial 模式的聚合函数在执行时会根据读入的原始数据更新对应的聚合缓冲区, 当处理完所有的输入数据后,返回的是聚合缓冲区中的中间数据 。 而 Final模式所起到的作用 是将聚合缓冲区的数据进行合并,然后返回最终的结果。 如下图所示,在最终分组计算总和 之前,可以先进行局部聚合处理,这样能够避免数据传输并减少计算量 。 因此,上述聚合过程 中在 map 阶段的 sum 函数处于 Partial模式,在 reduce 阶段的 sum 函数处于 Final模式 。

Complete模式和上述的 Partial/Final组合方式不一样,不进行局部聚合计算。 下图展示了同样的聚合函数采用 Complete模式的情形。 可以看到,最终阶段直接针对原始输入,中间没有局部聚合过程。 一般来讲, Complete模式应用在不支持Partial模式的聚合函数中。

相比 Partial、 Final和 Complete模式, PartialMerge模式的聚合函数主要是对聚合缓冲区进行合并,但此时仍然不是最终的结果。 ParitialMerge主要应用在 distinct语句中,如下图所示。 聚合语句针对同一张表进行 sum 和 count (distinct)查询,最终的执行过程包含了 4 步聚合操作。 第1步按照(A,C)分组,对 sum函数进行 Partial模式聚合计算;第2步是 PartialMerge模式,对上一步计算之后的聚合缓冲区进行合井,但此时仍然不是最终的结果;第3步分组的列 发生变化,再一次进行 Partial模式的 count计算;第4步完成 Final模式的最终计算。

Hive on Spark与SparkSQL的区别

Hive on Spark是由Cloudera发起,由Intel、MapR等公司共同参与的开源项目,其目的是把Spark作为Hive的一个计算引擎,将Hive的查询作为Spark的任务提交到Spark集群上进行计算。通过该项目,可以提高Hive查询的性能,同时为已经部署了Hive或者Spark的用户提供了更加灵活的选择,从而进一步提高Hive和Spark的普及率。

Hive on Spark是从Hive on MapReduce演进而来,Hive的整体解决方案很不错,但是从查询提交到结果返回需要相当长的时间,查询耗时太长,这个主要原因就是由于Hive原生是基于MapReduce的,那么如果我们不生成MapReduce Job,而是生成Spark Job,就可以充分利用Spark的快速执行能力来缩短HiveQL的响应时间。

Hive on Spark现在是Hive组件(从Hive1.1 release之后)的一部分。

SparkSQL和Hive On Spark都是在Spark上实现SQL的解决方案。Spark早先有Shark项目用来实现SQL层,不过后来推翻重做了,就变成了SparkSQL。这是Spark官方Databricks的项目,Spark项目本身主推的SQL实现。Hive On Spark比SparkSQL稍晚。Hive原本是没有很好支持MapReduce之外的引擎的,而Hive On Tez项目让Hive得以支持和Spark近似的Planning结构(非MapReduce的DAG)。所以在此基础上,Cloudera主导启动了Hive On Spark。这个项目得到了IBM,Intel和MapR的支持(但是没有Databricks)。

结构上Hive On Spark和SparkSQL都是一个翻译层,把一个SQL翻译成分布式可执行的Spark程序。比如一个SQL:

SELECT item_type, sum(price)
FROM item
GROUP item_type;

上面这个SQL脚本交给Hive或者类似的SQL引擎,它会“告诉”计算引擎做如下两个步骤:读取item表,抽出item_type,price这两个字段;对price计算初始的SUM(其实就是每个单独的price作为自己的SUM)因为GROUP BY说需要根据item_type分组,所以设定shuffle的key为item_type从第一组节点分组后分发给聚合节点,让相同的item_type汇总到同一个聚合节点,然后这些节点把每个组的Partial Sum再加在一起,就得到了最后结果。不管是Hive还是SparkSQL大致上都是做了上面这样的工作。

需要理解的是,Hive和SparkSQL都不负责计算,它们只是告诉Spark,你需要这样算那样算,但是本身并不直接参与计算。

Spark UDAF开发

分两种

  1. 无泛型约束的UDAF  extends UserDefinedAggregateFunction  extends Aggregator  dataframe设计的
  2. 有泛型约束的UDAF  extends Aggregator 该UDAF时允许添加泛型,保障函数更加安全。但是这种UDAF不可直接在SQL中被调用运算适用于强类型Datasets。

在Spark中使用
    1.编写UDAF<两种类型的UDAF都可以>
    2. 在spark中注册UDAF,为其绑定一个名字,使用

在Spark SQL 中使用
   1.编写UDAF<使用继承 UserDefinedAggregateFunction 类型编写>
   2. 打Jar包,并上传
   3. 注册临时聚合函数,并使用

ADD  jar TestSpark.jar;
CREATE  TEMPORARY FUNCTION  mean_my AS  'com.test.structure.udaf.MeanMy';
select t1.data,mean_my(t1.age)
from (
select 33 as age,  '1' as data 
union all 
select 55  as age, '1' as data 
union all 
select 66 as age, '2' as data
)t1
group by   t1.data;

自定义UDAF类

import org.apache.spark.sql.Row;
import org.apache.spark.sql.expressions.MutableAggregationBuffer;
import org.apache.spark.sql.expressions.UserDefinedAggregateFunction;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import java.util.ArrayList;
import java.util.List;public class MeanFloatUDAF extends UserDefinedAggregateFunction {/*** 聚合函数的输入数据结构*   函数的参数列表,不过需要写成StructType的格式*/@Overridepublic StructType inputSchema() {List<StructField> structFields = new ArrayList<>();structFields.add(DataTypes.createStructField( "field_nm", DataTypes.DoubleType, true ));return DataTypes.createStructType( structFields );
}/*** 聚缓存区数据结构 - 产生中间结果的数据类型* 如果是求平均数,存储总和以及计数,总和及计数就是中间结果* count    buffer.getInt(0)* sum_field   buffer.getDouble(1)*/
@Override
public StructType bufferSchema() {List<StructField> structFields = new ArrayList<>();structFields.add(DataTypes.createStructField( "count", DataTypes.IntegerType, true ));structFields.add(DataTypes.createStructField( "sum_field", DataTypes.DoubleType, true ));return DataTypes.createStructType( structFields );
}/*** 聚合函数返回值数据结构*/
@Override
public DataType dataType() {return DataTypes.DoubleType;
}/*** 聚合函数是否是幂等的,即相同输入是否总是能得到相同输出*/
@Override
public boolean deterministic() {return true;
}/*** 初始化缓冲区* buffer是中间结果,是Row类的子类*/
@Override
public void initialize(MutableAggregationBuffer buffer) {//相加的初始值,这里的要和上边的中间结果的类型和位置相对应 - buffer.getInt(0)buffer.update(0,0);//参与运算数字个数的初始值buffer.update(1,Double.valueOf(0.0) );
}/***  给聚合函数传入一条新数据进行处理*  //每有一条数据参与运算就更新一下中间结果(update相当于在每一个分区中的计算)*  buffer里面存放着累计的执行结果,input是当前的执行结果*/
@Override
public void update(MutableAggregationBuffer buffer, Row input) {//个数加1buffer.update(0,buffer.getInt(0)+1);//每有一个数字参与运算就进行相加(包含中间结果)buffer.update(1,buffer.getDouble(1)+Double.valueOf(input.getDouble(0)));
}/***  合并聚合函数缓冲区   //全局聚合*/
@Override
public void merge(MutableAggregationBuffer buffer1, Row buffer2) {buffer1.update(0,buffer1.getInt(0)+buffer2.getInt(0));buffer1.update(1,buffer1.getDouble(1)+buffer2.getDouble(1));
}/*** 计算最终结果*/
@Override
public Object evaluate(Row buffer) {return buffer.getDouble(1)/buffer.getInt(0);
}

使用自定义UDAF

//在Spark中使用  extends UserDefinedAggregateFunction类型的UDAF的使用import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import java.util.Arrays;
import java.util.List;public class MeanUDAFMain {
public static void main(String[] args){try {SparkSession spark = SparkSession.builder().appName("Java Spark SQL data sources example").config("spark.some.config.option", "some-value").master("local[2]").getOrCreate();List<Row> dataExample = Arrays.asList(RowFactory.create( "2019-0801", 4,9.2),RowFactory.create( "2020-0802", 3,8.6),RowFactory.create( "2021-0803",2,5.5),RowFactory.create( "2021-0803",2,5.5),RowFactory.create( "2021-0803",7,4.5));StructType schema = new StructType(new StructField[]{new StructField("date", DataTypes.StringType, false, Metadata.empty()),new StructField("dist_mem", DataTypes.IntegerType, false, Metadata.empty()),new StructField("dm_mem", DataTypes.DoubleType, false, Metadata.empty())});Dataset<Row> itemsDF = spark.createDataFrame(dataExample, schema);itemsDF.printSchema();itemsDF.createOrReplaceTempView("test_mean_table");// 注册自定义聚合函数 -2. 在spark中注册UDAF,为其绑定一个名字spark.udf().register("mymean",new MeanFloatUDAF ());spark.sql("select dist_mem  from test_mean_table").show();spark.sql("select date,mymean(dm_mem) memdoubleMean from test_mean_table group by date").show();} catch (Exception e) {e.printStackTrace();}
}
}

Hive UDAF开发

UDF、UDAF、UDTF需要实现的方法

类型方法
UDF

类:

GenericUDF


包路径:
org.apache.hadoop.hive.ql.udf.generic

initialize:类型检查,返回结果类型
入参:ObjectInspector[]
出参:ObjectInspector
 

evaluate:功能逻辑实现

入参:DeferredObject[]

出参:Object

getDisplayString:函数名称
入参:String[]

出参:String
 

close:关闭函数,释放资源等
入参:无

出参:void

UDTF

类:
GenericUDTF

包路径:
org.apache.hadoop.hive.ql.udf.generic

initialize:类型检查,返回结果类型
入参:StructObjectInspector
出参:StructObjectInspector

process:功能逻辑实现
**调用forward输出一行数据,可多次调用

入参:Object[]

出参:void

close:关闭函数,释放资源等
入参:无

出参:void

UDAF

类:
AbstractGenericUDAFResolver

包路径:
org.apache.hadoop.hive.ql.udf.generic

类:
GenericUDAFEvaluator

包路径:
org.apache.hadoop.hive.ql.udf.generic
 


类:

AbstractAggregationBuffer

包路径:
org.apache.hadoop.hive.ql.udf.generic

-----AbstractGenericUDAFResolver-----

getEvaluator:获取计算器
入参:TypeInfo[]
出参:GenericUDAFEvaluator

---------GenericUDAFEvaluator----------

init:
入参:Mode,ObjectInspector[]
出参:ObjectInspector

getNewAggregationBuffer:

入参:无

出参:AggregationBuffer

reset:

入参:AggregationBuffer

出参:void

iterate:

入参:AggregationBuffer,Object[]

出参:void

merge:

入参:AggregationBuffer,Object

出参:void


terminate:

入参:AggregationBuffer

出参:Object

terminatePartial:

入参:AggregationBuffer

出参:Object

--------AbstractAggregationBuffer-------
estimate:评估内存占用大小

入参:无

出参:int

UDAF说明

一个Buffer作为中间处理数据的缓冲:获取getNewAggregationBuffer、重置reset
四个模式(Mode):

  1. PARTIAL1:
    from original data to partial aggregation data:
    iterate() and terminatePartial() will be called.
  2. PARTIAL2:
    from partial aggregation data to partial aggregation data:
    merge() and terminatePartial() will be called.
  3. FINAL:
    from partial aggregation to full aggregation:
    merge() and terminate() will be called.
  4. COMPLETE:
    from original data directly to full aggregation:
    iterate() and terminate() will be called.

五个方法

  1. 初始化init
  2. 遍历iterate:PARTIAL1和COMPLETE阶段
  3. 合并merge:PARTIAL2和FINAL阶段
  4. 终止terminatePartial:PARTIAL1和PARTIAL2阶段
  5. terminate:COMPLETE和FINAL阶段

 

这篇关于【Spark精讲】一文讲透SparkSQL聚合过程以及UDAF开发的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/555314

相关文章

浅析Spring Security认证过程

类图 为了方便理解Spring Security认证流程,特意画了如下的类图,包含相关的核心认证类 概述 核心验证器 AuthenticationManager 该对象提供了认证方法的入口,接收一个Authentiaton对象作为参数; public interface AuthenticationManager {Authentication authenticate(Authenti

这15个Vue指令,让你的项目开发爽到爆

1. V-Hotkey 仓库地址: github.com/Dafrok/v-ho… Demo: 戳这里 https://dafrok.github.io/v-hotkey 安装: npm install --save v-hotkey 这个指令可以给组件绑定一个或多个快捷键。你想要通过按下 Escape 键后隐藏某个组件,按住 Control 和回车键再显示它吗?小菜一碟: <template

Hadoop企业开发案例调优场景

需求 (1)需求:从1G数据中,统计每个单词出现次数。服务器3台,每台配置4G内存,4核CPU,4线程。 (2)需求分析: 1G / 128m = 8个MapTask;1个ReduceTask;1个mrAppMaster 平均每个节点运行10个 / 3台 ≈ 3个任务(4    3    3) HDFS参数调优 (1)修改:hadoop-env.sh export HDFS_NAMENOD

作业提交过程之HDFSMapReduce

作业提交全过程详解 (1)作业提交 第1步:Client调用job.waitForCompletion方法,向整个集群提交MapReduce作业。 第2步:Client向RM申请一个作业id。 第3步:RM给Client返回该job资源的提交路径和作业id。 第4步:Client提交jar包、切片信息和配置文件到指定的资源提交路径。 第5步:Client提交完资源后,向RM申请运行MrAp

嵌入式QT开发:构建高效智能的嵌入式系统

摘要: 本文深入探讨了嵌入式 QT 相关的各个方面。从 QT 框架的基础架构和核心概念出发,详细阐述了其在嵌入式环境中的优势与特点。文中分析了嵌入式 QT 的开发环境搭建过程,包括交叉编译工具链的配置等关键步骤。进一步探讨了嵌入式 QT 的界面设计与开发,涵盖了从基本控件的使用到复杂界面布局的构建。同时也深入研究了信号与槽机制在嵌入式系统中的应用,以及嵌入式 QT 与硬件设备的交互,包括输入输出设

OpenHarmony鸿蒙开发( Beta5.0)无感配网详解

1、简介 无感配网是指在设备联网过程中无需输入热点相关账号信息,即可快速实现设备配网,是一种兼顾高效性、可靠性和安全性的配网方式。 2、配网原理 2.1 通信原理 手机和智能设备之间的信息传递,利用特有的NAN协议实现。利用手机和智能设备之间的WiFi 感知订阅、发布能力,实现了数字管家应用和设备之间的发现。在完成设备间的认证和响应后,即可发送相关配网数据。同时还支持与常规Sof

活用c4d官方开发文档查询代码

当你问AI助手比如豆包,如何用python禁止掉xpresso标签时候,它会提示到 这时候要用到两个东西。https://developers.maxon.net/论坛搜索和开发文档 比如这里我就在官方找到正确的id描述 然后我就把参数标签换过来

【机器学习】高斯过程的基本概念和应用领域以及在python中的实例

引言 高斯过程(Gaussian Process,简称GP)是一种概率模型,用于描述一组随机变量的联合概率分布,其中任何一个有限维度的子集都具有高斯分布 文章目录 引言一、高斯过程1.1 基本定义1.1.1 随机过程1.1.2 高斯分布 1.2 高斯过程的特性1.2.1 联合高斯性1.2.2 均值函数1.2.3 协方差函数(或核函数) 1.3 核函数1.4 高斯过程回归(Gauss

Linux_kernel驱动开发11

一、改回nfs方式挂载根文件系统         在产品将要上线之前,需要制作不同类型格式的根文件系统         在产品研发阶段,我们还是需要使用nfs的方式挂载根文件系统         优点:可以直接在上位机中修改文件系统内容,延长EMMC的寿命         【1】重启上位机nfs服务         sudo service nfs-kernel-server resta

【区块链 + 人才服务】区块链集成开发平台 | FISCO BCOS应用案例

随着区块链技术的快速发展,越来越多的企业开始将其应用于实际业务中。然而,区块链技术的专业性使得其集成开发成为一项挑战。针对此,广东中创智慧科技有限公司基于国产开源联盟链 FISCO BCOS 推出了区块链集成开发平台。该平台基于区块链技术,提供一套全面的区块链开发工具和开发环境,支持开发者快速开发和部署区块链应用。此外,该平台还可以提供一套全面的区块链开发教程和文档,帮助开发者快速上手区块链开发。