Spark SQL 用户自定义函数UDF、用户自定义聚合函数UDAF 教程(Java踩坑教学版)

简介:

在Spark中,也支持Hive中的自定义函数。自定义函数大致可以分为三种:

  • UDF(User-Defined-Function),即最基本的自定义函数,类似to_char,to_date等
  • UDAF(User- Defined Aggregation Funcation),用户自定义聚合函数,类似在group by之后使用的sum,avg等
  • UDTF(User-Defined Table-Generating Functions),用户自定义生成函数,有点像stream里面的flatMap

本篇就手把手教你如何编写UDF和UDAF

先来个简单的UDF

场景:
我们有这样一个文本文件:

1^^d
2^b^d
3^c^d
4^^d

在读取数据的时候,第二列的数据如果为空,需要显示'null',不为空就直接输出它的值。定义完成后,就可以直接在SparkSQL中使用了。

代码为:

package test;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
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;

/**
 * Created by xinghailong on 2017/2/23.
 */
public class test3 {
    public static void main(String[] args) {
        //创建spark的运行环境
        SparkConf sparkConf = new SparkConf();
        sparkConf.setMaster("local[2]");
        sparkConf.setAppName("test-udf");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        SQLContext sqlContext = new SQLContext(sc);
        //注册自定义方法
        sqlContext.udf().register("isNull", (String field,String defaultValue)->field==null?defaultValue:field, DataTypes.StringType);
        //读取文件
        JavaRDD<String> lines = sc.textFile( "C:\\test-udf.txt" );
        JavaRDD<Row> rows = lines.map(line-> RowFactory.create(line.split("\\^")));

        List<StructField> structFields = new ArrayList<StructField>();
        structFields.add(DataTypes.createStructField( "a", DataTypes.StringType, true ));
        structFields.add(DataTypes.createStructField( "b", DataTypes.StringType, true ));
        structFields.add(DataTypes.createStructField( "c", DataTypes.StringType, true ));
        StructType structType = DataTypes.createStructType( structFields );

        DataFrame test = sqlContext.createDataFrame( rows, structType);
        test.registerTempTable("test");
        
        sqlContext.sql("SELECT con_join(c,b) FROM test GROUP BY a").show();
        sc.stop();
    }
}

输出内容为:

+---+----+---+
|  a| _c1|  c|
+---+----+---+
|  1|null|  d|
|  2|   b|  d|
|  3|   c|  d|
|  4|null|  d|
+---+----+---+

其中比较关键的就是这句:

sqlContext.udf().register("isNull", (String field,String defaultValue)->field==null?defaultValue:field, DataTypes.StringType);

这里我直接用的java8的语法写的,如果是java8之前的版本,需要使用Function2创建匿名函数。

再来个自定义的UDAF—求平均数

先来个最简单的UDAF,求平均数。类似这种的操作有很多,比如最大值,最小值,累加,拼接等等,都可以采用相同的思路来做。

首先是需要定义UDAF函数

package test;

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;

/**
 * Created by xinghailong on 2017/2/23.
 */
public class MyAvg extends UserDefinedAggregateFunction {

    @Override
    public StructType inputSchema() {
        List<StructField> structFields = new ArrayList<>();
        structFields.add(DataTypes.createStructField( "field1", DataTypes.StringType, true ));
        return DataTypes.createStructType( structFields );
    }

    @Override
    public StructType bufferSchema() {
        List<StructField> structFields = new ArrayList<>();
        structFields.add(DataTypes.createStructField( "field1", DataTypes.IntegerType, true ));
        structFields.add(DataTypes.createStructField( "field2", DataTypes.IntegerType, true ));
        return DataTypes.createStructType( structFields );
    }

    @Override
    public DataType dataType() {
        return DataTypes.IntegerType;
    }

    @Override
    public boolean deterministic() {
        return false;
    }

    @Override
    public void initialize(MutableAggregationBuffer buffer) {
        buffer.update(0,0);
        buffer.update(1,0);
    }

    @Override
    public void update(MutableAggregationBuffer buffer, Row input) {
        buffer.update(0,buffer.getInt(0)+1);
        buffer.update(1,buffer.getInt(1)+Integer.valueOf(input.getString(0)));
    }

    @Override
    public void merge(MutableAggregationBuffer buffer1, Row buffer2) {
        buffer1.update(0,buffer1.getInt(0)+buffer2.getInt(0));
        buffer1.update(1,buffer1.getInt(1)+buffer2.getInt(1));
    }

    @Override
    public Object evaluate(Row buffer) {
        return buffer.getInt(1)/buffer.getInt(0);
    }
}

使用的时候,需要先注册,然后在spark sql里面就可以直接使用了:

package test;

import com.tgou.standford.misdw.udf.MyAvg;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
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;

/**
 * Created by xinghailong on 2017/2/23.
 */
public class test4 {
    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf();
        sparkConf.setMaster("local[2]");
        sparkConf.setAppName("test");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        SQLContext sqlContext = new SQLContext(sc);

        sqlContext.udf().register("my_avg",new MyAvg());

        JavaRDD<String> lines = sc.textFile( "C:\\test4.txt" );
        JavaRDD<Row> rows = lines.map(line-> RowFactory.create(line.split("\\^")));

        List<StructField> structFields = new ArrayList<StructField>();
        structFields.add(DataTypes.createStructField( "a", DataTypes.StringType, true ));
        structFields.add(DataTypes.createStructField( "b", DataTypes.StringType, true ));
        StructType structType = DataTypes.createStructType( structFields );

        DataFrame test = sqlContext.createDataFrame( rows, structType);
        test.registerTempTable("test");

        sqlContext.sql("SELECT my_avg(b) FROM test GROUP BY a").show();

        sc.stop();
    }
}

计算的文本内容为:

a^3
a^6
b^2
b^4
b^6

再来个无所不能的UDAF

真正的业务场景里面,总会有千奇百怪的需求,比如:

  • 想要按照某个字段分组,取其中的一个最大值
  • 想要按照某个字段分组,对分组内容的数据按照特定字段统计累加
  • 想要按照某个字段分组,针对特定的条件,拼接字符串

再比如一个场景,需要按照某个字段分组,然后分组内的数据,又需要按照某一列进行去重,最后再计算值

  • 1 按照某个字段分组
  • 2 分组校验条件
  • 3 然后处理字段

如果不用UDAF,你要是写spark可能需要这样做:

rdd.groupBy(r->r.xxx)
    .map(t2->{
        HashSet<String> set = new HashSet<>();
        for(Object p : t2._2){
            if(p.getBs() > 0 ){
                map.put(xx,yyy)
            }
        }
        return StringUtils.join(set.toArray(),",");
    });

上面是一段伪码,不保证正常运行哈。

这样写,其实也能应付需求了,但是代码显得略有点丑陋。还是不如SparkSQL看的清晰明了...

所以我们再尝试用SparkSql中的UDAF来一版!

首先需要创建UDAF类

import org.apache.commons.lang.StringUtils;
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.*;

import java.util.*;

/**
 *
 * Created by xinghailong on 2017/2/23.
 */
public class ConditionJoinUDAF extends UserDefinedAggregateFunction {
    @Override
    public StructType inputSchema() {
        List<StructField> structFields = new ArrayList<>();
        structFields.add(DataTypes.createStructField( "field1", DataTypes.IntegerType, true ));
        structFields.add(DataTypes.createStructField( "field2", DataTypes.StringType, true ));
        return DataTypes.createStructType( structFields );
    }

    @Override
    public StructType bufferSchema() {
        List<StructField> structFields = new ArrayList<>();
        structFields.add(DataTypes.createStructField( "field", DataTypes.StringType, true ));
        return DataTypes.createStructType( structFields );
    }

    @Override
    public DataType dataType() {
        return DataTypes.StringType;
    }

    @Override
    public boolean deterministic() {//是否强制每次执行的结果相同
        return false;
    }

    @Override
    public void initialize(MutableAggregationBuffer buffer) {//初始化
        buffer.update(0,"");
    }

    @Override
    public void update(MutableAggregationBuffer buffer, Row input) {//相同的executor间的数据合并
        Integer bs = input.getInt(0);
        String field = buffer.getString(0);
        String in = input.getString(1);
        if(bs > 0 && !"".equals(in) && !field.contains(in)){
            field += ","+in;
        }
        buffer.update(0,field);
    }

    @Override
    public void merge(MutableAggregationBuffer buffer1, Row buffer2) {//不同excutor间的数据合并
        String field1 = buffer1.getString(0);
        String field2 = buffer2.getString(0);
        if(!"".equals(field2)){
            field1 += ","+field2;
        }
        buffer1.update(0,field1);
    }

    @Override
    public Object evaluate(Row buffer) {//根据Buffer计算结果
        return StringUtils.join(Arrays.stream(buffer.getString(0).split(",")).filter(line->!line.equals("")).toArray(),",");
    }
}

拿一个例子坐下实验:

a^1111^2
a^1111^2
a^1111^2
a^1111^2
a^1111^2
a^2222^0
a^3333^1
b^4444^0
b^5555^3
c^6666^0

按照第一列进行分组,不同的第三列值,进行拼接。

package test;

import test.ConditionJoinUDAF;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
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;

/**
 * Created by xinghailong on 2017/2/23.
 */
public class test2 {
    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf();
        sparkConf.setMaster("local[2]");
        sparkConf.setAppName("test");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        SQLContext sqlContext = new SQLContext(sc);

        sqlContext.udf().register("con_join",new ConditionJoinUDAF());

        JavaRDD<String> lines = sc.textFile( "C:\\test2.txt" );
        JavaRDD<Row> rows = lines.map(line-> RowFactory.create(line.split("\\^")));

        List<StructField> structFields = new ArrayList<StructField>();
        structFields.add(DataTypes.createStructField( "a", DataTypes.StringType, true ));
        structFields.add(DataTypes.createStructField( "b", DataTypes.StringType, true ));
        structFields.add(DataTypes.createStructField( "c", DataTypes.StringType, true ));
        StructType structType = DataTypes.createStructType( structFields );

        DataFrame test = sqlContext.createDataFrame( rows, structType);
        test.registerTempTable("test");

        sqlContext.sql("SELECT con_join(c,b) FROM test GROUP BY a").show();

        sc.stop();
    }

}

这样SQL简洁明了,就能表达意思了。

参考

本文转自博客园xingoo的博客,原文链接:Spark SQL 用户自定义函数UDF、用户自定义聚合函数UDAF 教程(Java踩坑教学版),如需转载请自行联系原博主。
相关文章
|
5月前
|
SQL 存储 Oracle
sql数据库使用教程
SQL(Structured Query Language)结构化查询语言是一种用于操作数据库的标准语言,被广泛应用于关系型数据库管理系统(RDBMS),如MySQL、Oracle、Microsoft
|
1月前
|
消息中间件 存储 分布式计算
大数据-103 Spark Streaming Kafka Offset管理详解 Scala自定义Offset
大数据-103 Spark Streaming Kafka Offset管理详解 Scala自定义Offset
82 0
|
12天前
|
SQL 缓存 Java
【详细实用のMyBatis教程】获取参数值和结果的各种情况、自定义映射、动态SQL、多级缓存、逆向工程、分页插件
本文详细介绍了MyBatis的各种常见用法MyBatis多级缓存、逆向工程、分页插件 包括获取参数值和结果的各种情况、自定义映射resultMap、动态SQL
【详细实用のMyBatis教程】获取参数值和结果的各种情况、自定义映射、动态SQL、多级缓存、逆向工程、分页插件
|
29天前
|
SQL 数据管理 数据库
SQL语句实例教程:掌握数据查询、更新与管理的关键技巧
SQL(Structured Query Language,结构化查询语言)是数据库管理和操作的核心工具
|
5月前
|
SQL 数据库
sql数据库教程设计
SQL数据库教程设计可以分为以下几个步骤: 1. 确定教学目标:首先,需要明确教程的教学目标,例如让学生掌握SQL语言的基础知识,包括数据查询、数据操作、数据定义等,以及培养学生的实际操作能力,
|
2月前
|
SQL 安全 Go
SQL注入不可怕,XSS也不难防!Python Web安全进阶教程,让你安心做开发!
在Web开发中,安全至关重要,尤其要警惕SQL注入和XSS攻击。SQL注入通过在数据库查询中插入恶意代码来窃取或篡改数据,而XSS攻击则通过注入恶意脚本来窃取用户敏感信息。本文将带你深入了解这两种威胁,并提供Python实战技巧,包括使用参数化查询和ORM框架防御SQL注入,以及利用模板引擎自动转义和内容安全策略(CSP)防范XSS攻击。通过掌握这些方法,你将能够更加自信地应对Web安全挑战,确保应用程序的安全性。
85 3
|
1月前
|
缓存 分布式计算 大数据
大数据-90 Spark 集群 RDD 编程-高阶 RDD容错机制、RDD的分区、自定义分区器(Scala编写)、RDD创建方式(一)
大数据-90 Spark 集群 RDD 编程-高阶 RDD容错机制、RDD的分区、自定义分区器(Scala编写)、RDD创建方式(一)
45 0
|
1月前
|
分布式计算 算法 大数据
大数据-90 Spark 集群 RDD 编程-高阶 RDD容错机制、RDD的分区、自定义分区器(Scala编写)、RDD创建方式(二)
大数据-90 Spark 集群 RDD 编程-高阶 RDD容错机制、RDD的分区、自定义分区器(Scala编写)、RDD创建方式(二)
50 0
spark3.5.1中内置函数大全
spark3.5.1中内置函数大全
|
4月前
|
SQL 安全 Go
SQL注入不可怕,XSS也不难防!Python Web安全进阶教程,让你安心做开发!
【7月更文挑战第26天】在 Web 开发中, SQL 注入与 XSS 攻击常令人担忧, 但掌握正确防御策略可化解风险. 对抗 SQL 注入的核心是避免直接拼接用户输入至 SQL 语句. 使用 Python 的参数化查询 (如 sqlite3 库) 和 ORM 框架 (如 Django, SQLAlchemy) 可有效防范. 防范 XSS 攻击需严格过滤及转义用户输入. 利用 Django 模板引擎自动转义功能, 或手动转义及设置内容安全策略 (CSP) 来增强防护. 掌握这些技巧, 让你在 Python Web 开发中更加安心. 安全是个持续学习的过程, 不断提升才能有效保护应用.
54 1