一、SparkSQL概述

1.1 什么是SparkSQL

Spark是用于结构化数据处理的Spark模块。与基本的Spark RDD API不同，SparkSQL提供的接口为Spark提供了有关数据结构和正在执行的计算的更多信息。在内部，SparkSQL使用这些额外的信息来执行额外的优化。与SparkSQL交互的方式有很多种，包括SQL和DatasetAPI。结算时，使用相同的执行引擎，与你用于表计算的API/语言无关。

它提供了⼀个编程抽象叫做DataFrame/Dataset，它可以理解为⼀个基于RDD数据模型的更⾼级数据
模型，带有结构化元信息（schema），DataFrame其实就是Dataset[Row]，Spark SQL可以将针对
DataFrame/Dataset的各类SQL运算，翻译成RDD的各类算⼦执⾏计划，从⽽⼤⼤简化数据运算编程
（请联想Hive）

1.2 为什么要有SparkSQL

SparkSQL的发展

1、发展历史

RDD（Spark1.0）=> Dateframe（Spark1.3） =>Dataset（Spark1.6）

如果同样的数据都给到这三个数据结构，它们分别计算之后，都会给出相同的结果。

不同的是它们执行效率和执行方式。在现在的版本中，dataset性能最好，已经成为了唯一使用的接口。其中Dataframe已经在底层被看作是特殊泛型的DataSet。

2、三者的共性

（1）RDD、DataFrame、DataSet全都是Spark平台下的分布式弹性数据集，为处理大型数据通过便利。

（2）三者都有惰性机制，在进行创建、转换，如map方法时，不会立即执行，只有在遇到Action行动算子实，三者才会开始遍历运算。

（3）三者有许多共同的函数，例如filter，sortby等

（4）三者都会根据Spark的内存情况自动缓存运算。

（5）三者都有分区的概念

SparkSQL的特点

1、易整合：无缝的整合了SQL查询和Spark编程

2、统一的数据访问方式：使用相同的方式连接不同的数据源

3、兼容Hive：在已有的仓库上直接运行SQL或者HQL

4、标准的数据连接：通过JDBC或者ODBC来连接

二、SparkSQL 编程

2.1 SparkSession 新的起始点

在老的版本中，SparkSQL提供两种SQL查询起始点：

（1）一个叫SQLContext，用于Spark自己提供的SQL查询；

（2）一个叫HiveContext，用于连接Hive的查询。

SparkSession是Spark最新的SQL查询起始点，实质上是SQLContext和HiveContext的组合，所以在SQLContext和HiveContext上可用的API在SparkSession上同样是可以使用的。

SparkSession内部封装了SparkContext，所以计算实际上是由SparkContext完成的。当我们使用spark-shell的时候，Spark框架会自动的创建一个名称叫做Spark的SparkSession，就像我们以前可以自动获取到一个sc来表示SparkContext。

从JSON⽂件加载DataFrame

package org.example

import org.apache.spark.sql.{DataFrame, SparkSession}

object S04_DataFrame读取复杂json文件 {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder()
      .appName("json创建dataframe")
      .master("local")
      .config("spark.default.parallelism", 20)
      .getOrCreate()
    import spark.implicits._
    val df: DataFrame = spark.read.json("F:\\代码区\\sparkp\\datas\\c.txt")
    df.show(100,false)
    df.printSchema()
    df.createTempView("df1")
    //取每个人的mother
    spark.sql(
      """
        |select
        |family[0].name
        |from df1
        |""".stripMargin).show(100,false)

    spark.close()
  }

}

package org.example

import org.apache.spark.sql.types.{DataType, DataTypes, StructField, StructType}
import org.apache.spark.sql.{DataFrame, SparkSession}

object S05_DataFrame读取复杂json文件 {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder()
      .appName("json创建dataframe")
      .master("local")
      .config("spark.default.parallelism", 20)
      .getOrCreate()


    //解析出来的结果比较丑陋,info被认为了struct类型,而struct类型的成员变量是统一的
    //手动指定schema来改善上面的问题
    val scheam=StructType(Seq(
      StructField("id",DataTypes.LongType),
      StructField("info",DataTypes.createMapType(DataTypes.StringType,DataTypes.StringType))
    ))
    val df: DataFrame = spark.read.schema(scheam).json("F:\\代码区\\sparkp\\datas\\d.txt")
    df.show(100, false)
    df.printSchema()
        //找出有年龄属性的数据后，求平均值
    df.createTempView("f4")
    spark.sql(
      """
        |select
        |avg(info['age']) as avg_age
        |from f4
        |where info['age'] is not null
        |""".stripMargin).show(100,false)

    spark.close()
  }

}

从非结构化⽂件加载DataFrame

sparksql创建wordcount

package org.example

import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
/*
* sparksql将输入数据视做非结构化数据读的时候
* 就是把整行内容当成一个字段(value:String)
* */
object S06_普通文本文件 {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder()
      .appName("普通文本创建dataframe")
      .master("local")
      .config("spark.default.parallelism", 20)
      .config("spark.sql.shuffle.partitions",1)
      .getOrCreate()
    //dataframe就是dataset的一个特例:DataSet[Row]
    val df: DataFrame = spark.read.text("F:\\代码区\\sparkp\\datas\\f.txt")
    df.show(100,false)
    df.printSchema()
    val ds: Dataset[String] = spark.read.textFile("F:\\代码区\\sparkp\\datas\\f.txt")
    ds.show(100,false)
    ds.printSchema()
    //wordcount
    df.createTempView("ff")
    val words=spark.sql(
      """
        |select
        |words,count(1) as cnt
        |from
        |(select
        | explode(split(value,'\\s+')) as words
        | from ff)o
        | group by words
        |""".stripMargin
    )
    words.show(100,false)
    words.printSchema()
    import spark.implicits._
    val ds1:DataFrame = df.flatMap(
      row => {
        val line: String = row.getAs[String]("value")
        line.split("\\s+")
      }
    ).toDF("word")
    ds1.createTempView("ds1")
    spark.sql(
      """
        |select
        | word,count(1) as cnt
        | from df2
        | group by word
        |""".stripMargin
    ).show()




    spark.close()

  }

}

从Parquet⽂件进⾏创建

Parquet⽂件是⼀种列式存储⽂件格式，⽂件⾃带schema描述信息,自我描述

package org.example

import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.types.{DataTypes, StructField, StructType}

object S07_Parque文件 {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder()
      .appName("普通文本创建dataframe")
      .master("local")
      .config("spark.default.parallelism", 20)
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()
    val schema=StructType(
      Seq(
        StructField("id",DataTypes.LongType),
        StructField("country",DataTypes.StringType),
        StructField("name",DataTypes.StringType),
        StructField("battle",DataTypes.DoubleType),
        StructField("age",DataTypes.LongType)
      )
    )
    val df: DataFrame = spark.read.option("header", "true").schema(schema).csv("F:\\代码区\\sparkp\\datas\\a.txt")
    df.write.parquet("datas/parquet/")
    df.write.parquet("data/orc/")
   //读取上面job生成的parque文件
   val df2: DataFrame = spark.read.parquet("datas/parquet")
    df2.show(100,false)
    df2.printSchema()
    //读取上面job生成job生成的orc文件 //orc也是自我描述的列式存储文件格式
    val df3: DataFrame = spark.read.parquet("data/orc")
    df3.show(100,false)
    df3.printSchema()
    spark.close()
  }

}

外部存储服务创建DF

（1）从JDBC连接数据库服务器进⾏创建dataframe

package org.example

import org.apache.spark.sql.{DataFrameReader, SaveMode, SparkSession}

object Spark09_Mysql {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder()
      .appName("创建JDBC")
      .master("local")
      .config("spark.default.parallelism", 20)
      .getOrCreate()
    val df= spark.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/atguigudb")
      .option("driver", "com.mysql.jdbc.Driver")
      .option("user", "root")
      .option("password", "915425")
      .option("dbtable", "user").load()
//      .load().show
    //保存数据
    df.write.format("jdbc")
      .option("url","jdbc:mysql://localhost:3306/atguigudb")
      .option("driver", "com.mysql.jdbc.Driver")
      .option("user", "root")
      .option("password", "915425")
      .option("dbtable", "user1")
      .mode(SaveMode.Append).save()
    spark.stop()
  }

}

val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL")

//创建SparkSession对象
val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()

import spark.implicits._

//方式1：通用的load方法读取
spark.read.format("jdbc")
  .option("url", "jdbc:mysql://linux1:3306/spark-sql")
  .option("driver", "com.mysql.jdbc.Driver")
  .option("user", "root")
  .option("password", "123123")
  .option("dbtable", "user")
  .load().show


//方式2:通用的load方法读取 参数另一种形式
spark.read.format("jdbc")
  .options(Map("url"->"jdbc:mysql://linux1:3306/spark-sql?user=root&password=123123",
    "dbtable"->"user","driver"->"com.mysql.jdbc.Driver")).load().show

//方式3:使用jdbc方法读取
val props: Properties = new Properties()
props.setProperty("user", "root")
props.setProperty("password", "123123")
val df: DataFrame = spark.read.jdbc("jdbc:mysql://linux1:3306/spark-sql", "user", props)
df.show

//释放资源

写入数据

case class User2(name: String, age: Long)
。。。
val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL")
 
//创建SparkSession对象
val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
import spark.implicits._
 
val rdd: RDD[User2] = spark.sparkContext.makeRDD(List(User2("lisi", 20), User2("zs", 30)))
val ds: Dataset[User2] = rdd.toDS
//方式1：通用的方式  format指定写出类型
 
//方式2：通过jdbc方法
val props: Properties = new Properties()
props.setProperty("user", "root")
props.setProperty("password", "123123")
ds.write.mode(SaveMode.Append).jdbc("jdbc:mysql://linux1:3306/spark-sql", "user", props)
 
//释放资源
spark.stop()

spark整合hive原理--访问hive元数据服务原理

bin/hive --service metastore 1>/dev/null 2>&1 &

既然具备了hive的功能，那么就可以执⾏⼀切hive中能执⾏的动作：
◦ 建表
◦ show 表
◦ 建库
◦ show 库
◦ alter表
◦ ……
只不过，此时看⻅的表是spark中集成的hive的本地元数据库中的表！
如果想让spark中集成的hive，看⻅你外部集群中的hive的表，只要修改配置：把spark端的hive的元
数据服务地址，指向外部集群中hive的元数据服务地址；
有两种指定办法：
◦ 在spark端加⼊hive-site.xml ，⾥⾯配置 ⽬标元数据库 mysql的连接信息
这会使得spark中集成的hive直接访问mysql元数据库
◦ 在spark端加⼊hive-site.xml ，⾥⾯配置 ⽬标hive的元数据服务器地址
这会使得spark中集成的hive通过外部独⽴的hive元数据服务来访问元数据库

#在hive中创建表
use  atguigudb
CREATE TABLE user (
    id int,
    name VARCHAR(50) NOT NULL,
    age INT  
)
row format delimited fields terminated by ','
stored as textfile
tblproperties(
"external"="true");
insert into user(id,name,age)
values (1,'sun',60),
(2,'jie',80),
(6,'ss',90);
#加载数据
load data local inpath '/root/x.txt' into table t_sparkset;

package org.example

import org.apache.spark.sql.SparkSession

object Spark10_创建hive表 {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession
    .builder()
    .appName(this.getClass.getSimpleName)
    .master("local[*]")
      .config("hive.metastore.uris","hdfs://node1:9083")
     // 启⽤hive⽀持,需要调⽤enableHiveSupport，还需要添加⼀个依赖 spark-hive
    // 默认sparksql内置了⾃⼰的hive
    // 如果程序能从classpath中加载到hive-site配置⽂件，那么它访问的hive元数据库就不是本地内
    // 如果程序能从classpath中加载到core-site配置⽂件，那么它访问的⽂件系统也不再是本地⽂件
    .enableHiveSupport()
    .getOrCreate()
    val res=spark.sql(
      """
        |select * from default.t_sparktset
        |""".stripMargin
    )
    res.show(100,false)
    //读取hive的分区表 并指定要读取的分区
    spark.sql(""" select * from t_acc_log where dt='2021-12-03' """).show(100,false)
    spark.read.table("t_acc_log").where("dt='2021-12-04'").show(100,false)
    spark.close()

  }

}

用户自定义函数

强类型的Dataset和弱类型的DataFrame都提供了相关的聚合函数，如 count()，countDistinct()，avg()，max()，min()。除此之外，用户可以设定自己的自定义聚合函数。通过继承UserDefinedAggregateFunction来实现用户自定义弱类型聚合函数。从Spark3.0版本后，UserDefinedAggregateFunction已经不推荐使用了。可以统一采用强类型聚合函数

package org.example
import org.apache.spark.sql.{DataFrame, SparkSession}
object S11_SPARKSQL的UDF自定义函数 {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder()
      .appName("自定义函数demo")
      .master("local")
      .getOrCreate()


    val df: DataFrame = spark.read
      .option("header", "true")
      .csv("file:///F:\\代码区\\sparkp\\datas\\a.txt")
      .toDF("id", "country", "name", "battle", "age")

    df.createTempView("df")

    val func = (c: String, n: String) => {
      val firstName: String = n.substring(0, 1)
      val lastName: String = n.substring(1)
      firstName + c + lastName
    }

    // 往sparksql的catalog中，注册函数名
    spark.udf.register("qiguai", func)


    // id,country,name,battle,age
    val res: DataFrame = spark.sql(
      """
        |select
        |  id,country,name,battle,age,qiguai(country,name) as new_name
        |from df
        |
        |""".stripMargin)

    res.show(100, false)


    spark.close()

  }

}

自定义UDF实战案例

package org.example
import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
import org.apache.spark.sql.{DataFrame, SparkSession}
object S11_SPARKQL的UDF自定义函数应用实战 {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder()
      .appName("自定义函数demo")
      .master("local")
      .getOrCreate()

    val schema = StructType(Seq(
      StructField("id", DataTypes.IntegerType),
      StructField("f1", DataTypes.IntegerType),
      StructField("f2", DataTypes.IntegerType),
      StructField("f3", DataTypes.IntegerType),
      StructField("gender", DataTypes.StringType),
    ))

    val sample: DataFrame = spark.read.schema(schema).csv("datas/stu/input/sample.txt")


    val schema2 = StructType(Seq(
      StructField("id", DataTypes.StringType),
      StructField("f1", DataTypes.IntegerType),
      StructField("f2", DataTypes.IntegerType),
      StructField("f3", DataTypes.IntegerType)
    ))
    val test: DataFrame = spark.read.schema(schema2).csv("datas/stu/input/test.txt")



    sample.createTempView("sample")
    test.createTempView("test")


    // 首先写一个普通的scala函数
    //接收两个数组，返回一个距离
    val dist = (arr1: Array[Int], arr2: Array[Int]) => {
      //欧式距离
      //每个测试点与距离点的距离
      arr1.zip(arr2).map(tp => Math.pow(tp._1 - tp._2, 2)).sum
    }

    spark.udf.register("dist", dist)

    val distDf = spark.sql(
      """
        |
        |select
        |
        |sample.id as sample_id,
        |sample.gender as sample_gender,
        |test.id,
        |dist(array(sample.f1,sample.f2,sample.f3),array(test.f1,test.f2,test.f3)) as dist
        |
        |from sample cross join test
        |
        |

        |""".stripMargin)
    distDf.createTempView("dist_df")


    // TODO 距离算好，后续逻辑纯sql可以解决：
    // TODO 找到每个测试人距离最近的3个样本人，看这3个洋本人中，哪种性别最多，结果就是这种性别
    /**
     * dist_df
     * +---------+-------------+---+------+
     * |sample_id|sample_gender|id |dist  |
     * +---------+-------------+---+------+
     * |1        |m            |a  |221.0 |
     * |1        |m            |b  |874.0 |
     * |2        |m            |a  |46.0  |
     * |2        |m            |b  |1389.0|
     * |3        |m            |a  |264.0 |
     * |3        |m            |b  |973.0 |
     * |4        |f            |a  |1406.0|
     * |4        |f            |b  |59.0  |
     * |5        |f            |a  |1668.0|
     * |5        |f            |b  |21.0  |
     * |6        |f            |a  |2001.0|
     * |6        |f            |b  |4.0   |
     * +---------+-------------+---+------+
     */
    spark.sql(
      """
        |select
        |  sample_id,
        |  sample_gender,
        |  id,
        |  rn
        |from (
        |select
        |  sample_id,
        |  sample_gender,
        |  id,
        |  row_number() over(partition by id order by dist) as rn
        |from dist_df ) o
        |where rn <=3

        |
        |
        |""".stripMargin).createTempView("knn")
    /**
     * knn
     * +---------+-------------+---+---+
     * |sample_id|sample_gender|id |rn |
     * +---------+-------------+---+---+
     * |6        |f            |b  |1  |
     * |5        |f            |b  |2  |
     * |4        |m            |b  |3  |
     * |2        |m            |a  |1  |
     * |1        |f            |a  |2  |
     * |3        |m            |a  |3  |
     * +---------+-------------+---+---+
     */

    val res = spark.sql(
      """
        |select
        |   id,
        |   if(sum(if(sample_gender='f',0,1))>=2,'male','female') as gender
        |from knn
        |group by id
        |
        |""".stripMargin)

    res.show(100, false)


    spark.close()

  }

}

java开发spark快速上手

package javaspark;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import scala.Tuple2;
import java.util.Arrays;
import java.util.Iterator;

public class JavaWordCount {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("java版wordcount");
        JavaSparkContext sc = new JavaSparkContext(conf);
        // 读数据
        JavaRDD<String> rdd = sc.textFile("F:\\代码区\\sparkp\\datas\\wordcount\\input\\a.txt");

        JavaRDD<String> rdd2 = rdd.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String s) throws Exception {
                return Arrays.stream(s.split("\\s+")).iterator();
            }
        });
        JavaPairRDD<String, Integer> rdd3 = rdd2.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String word) throws Exception {
                return new Tuple2<>(word, 1);
            }
        });
        JavaPairRDD<String, Integer> res = rdd3.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });
        res.foreach(new VoidFunction<Tuple2<String, Integer>>() {
            @Override
            public void call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                System.out.println(stringIntegerTuple2._1+","+stringIntegerTuple2._2);
            }
        });
//        res.foreach(new VoidFunction<Tuple2<String, Integer>>() {
//            @Override
//            public void call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
//                System.out.println(stringIntegerTuple2._1+""+stringIntegerTuple2._2);
//            }
//        });
        sc.stop();

    }
}

package javaspark;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
public class JavaWordCount2 {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("java版wordcount");

        JavaSparkContext sc = new JavaSparkContext(conf);

        // 读数据
        JavaPairRDD<String, Integer> res = sc.textFile("F:\\代码区\\sparkp\\datas\\wordcount\\input\\a.txt")
                .flatMap(s -> Arrays.asList(s.split("\\s+")).iterator())
                .mapToPair(w -> new Tuple2<>(w, 1))
                .reduceByKey((v1, v2) -> v1 + v2);

        List<Tuple2<String, Integer>> lst = res.collect();
        System.out.println(lst);

        sc.stop();
    }
}

自定义UDFA工作逻辑

UDAF - 弱类型

package org.example
import org.apache.spark.SparkConf
import org.apache.spark.sql.expressions.{Aggregator, MutableAggregationBuffer, UserDefinedAggregateFunction, UserDefinedFunction}
import org.apache.spark.sql._
import org.apache.spark.sql.types.{DataType, LongType, StructField, StructType}
object SparkSQL04_弱UDAF {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setAppName("SparkSQL").setMaster("local[*]")

    val spark: SparkSession =
      SparkSession.builder()
        //.appName("SparkSQL")
        //.master("local[*]")
        .config(sparkConf)
        .getOrCreate()
    val df: DataFrame = spark.read.json("datas/user.json")
    df.createOrReplaceTempView("user")
        spark.udf.register("ageAvg",new MyAvgAggregator())
    spark.sql("select ageAvg(age) from user").show()
    spark.close()

    //写代码不管用不用都导入。
    import spark.implicits._
  }
  class  MyAvgAggregator extends UserDefinedAggregateFunction{
    //输入数据结构
    override def inputSchema: StructType = {
      StructType(Array(
        StructField("age",LongType)
      ))
    }
    //缓存区数据的结构

    override def bufferSchema: StructType = {
      StructType(
        Array(
          StructField("total",LongType),
          StructField("count",LongType)
        )
      )
    }
    //输出：函数计算结果的数据类型
    override def dataType: DataType = LongType
    //函数的稳定性
    override def deterministic: Boolean = true
    //缓冲区初始化
    override def initialize(buffer: MutableAggregationBuffer): Unit =
      {
//        buffer(0)=0L
//        buffer(1)=0L
        buffer.update(0,0L)
        buffer.update(1,0L)
      }
    //根据输入的值更新缓冲区
    override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
      buffer.update(0,buffer.getLong(0)+input.getLong(0))
      buffer.update(1,buffer.getLong(1)+1)
    }
    //合并
    override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
      buffer1.update(0,buffer1.getLong(0)+buffer2.getLong(0))
      buffer1.update(1,buffer1.getLong(1)+buffer2.getLong(1))
    }
    //计算平均值
    override def evaluate(buffer: Row): Any = {
      buffer.getLong(0)/buffer.getLong(1)
    }
  }

}

UDAF - 强类型

package org.example
import org.apache.spark.SparkConf
import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql._
object SparkSQL04_UDAF {
  def main(args: Array[String]): Unit = {
    //新的起点: SparkSession
    //需求：计算平均工资

    val sparkConf: SparkConf = new SparkConf().setAppName("SparkSQL").setMaster("local[*]")

    val spark: SparkSession =
      SparkSession.builder()
        //.appName("SparkSQL")
        //.master("local[*]")
        .config(sparkConf)
        .getOrCreate()
    val df: DataFrame = spark.read.json("datas/user.json")
    df.createOrReplaceTempView("user")
    spark.udf.register("ageAvg",functions.udaf(new My_Avg_Aggregator()))
    spark.sql("select ageAvg(age) from user").show()

    //写代码不管用不用都导入。
//    import spark.implicits._
//
//    val df: DataFrame = spark.read.json("data/user.json")
//    val ds: Dataset[User] = df.as[User]
//
//    val my_Avg_Aggregator = new My_Avg_Aggregator
//    //将UDAF函数转化为查询的列对象
//    val column: TypedColumn[User, Double] = my_Avg_Aggregator.toColumn
//
//    ds.select(column).show()

    spark.stop()

  }

  case class User(id: Long, name: String, age: Long)

  case class AgeBuffer(var totalAge: Long, var totalCount: Long)

   /**
    * 自定义聚合函数类
   * 泛型:
   * IN: User
   * BUF: AgeBuffer
   * OUT: Double
   */
  class My_Avg_Aggregator extends Aggregator[Long, AgeBuffer, Long] {
     //初始值缓冲区的初始化
     override def zero: AgeBuffer = AgeBuffer(0L, 0L)

     //根据输入的数据更新缓冲区的数据
   

     override def reduce(b: AgeBuffer, a: Long): AgeBuffer = 
       {
         b.totalAge=b.totalAge+a
         b.totalCount=b.totalCount+1
         b
       }
       //合并缓冲区

     override def merge(b1: AgeBuffer, b2: AgeBuffer): AgeBuffer = {
       b1.totalAge=b1.totalAge+b2.totalAge
       b1.totalCount=b1.totalCount+b2.totalCount
       b1
     }
     //计算缓冲区

     override def finish(reduction: AgeBuffer): Long = {
       reduction.totalAge/reduction.totalCount
     }

     override def bufferEncoder: Encoder[AgeBuffer] = Encoders.product

     override def outputEncoder: Encoder[Long] = Encoders.scalaLong
   }

}

自定义UDFA实战案例

package org.example
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.types.{DataType, DataTypes, StructField, StructType}
import util.BitMapUtil
object S11_SPARKQL的UDF自定义函数应用实战1 {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder()
      .appName("自定义UDAF")
      .master("local")
      .config("spark.sql.shuffle.partitions", 2)
      .enableHiveSupport()
      .getOrCreate()


    // 加载待处理数据
    val schema = StructType(Seq(
      StructField("id", DataTypes.IntegerType),
      StructField("province", DataTypes.StringType),
      StructField("city", DataTypes.StringType),
      StructField("region", DataTypes.StringType),
      StructField("pv", DataTypes.IntegerType),
    ))
    val df: DataFrame = spark.read.schema(schema).csv("F:\\代码区\\sparkp\\datas\\input\\data.csv")
    df.createTempView("df")

    // 注册自定义UDAF
    import org.apache.spark.sql.functions.udaf
    spark.udf.register("gen_bitmap", udaf(BitMapGenUDA))
    spark.udf.register("merge_bitmap", udaf(BitMapOrMergeUDAF))

    val card = (bmBytes: Array[Byte]) => {
      BitMapUtil.deSerBitMap(bmBytes).getCardinality
    }
    spark.udf.register("card_bm", card)


    // 按省市区统计pv总数和uv总数并保存到hive中
    val pcrReport = spark.sql(
      """
        |
        |select
        |province,
        |city,
        |region,
        |sum(pv) as pv_amt,
        |card_bm(gen_bitmap(id)) as uv_cnt,
        |gen_bitmap(id) as bitmap
        |
        |from df
        |group by province,city,region
        |
        |
        |""".stripMargin)
    pcrReport.write.saveAsTable("pcr_report")


    // 读hive中省市区报表，聚合出省市报表
    spark.sql(
      """
        |
        |select
        |province,
        |city,
        |sum(pv_amt)  as pv_amt,
        |card_bm(merge_bitmap(bitmap)) as uv_cnt,
        |merge_bitmap(bitmap) as bitmap
        |
        |from pcr_report
        |group by province,city
        |
        |""".stripMargin).show(100, false)


    // 读hive中省市区报表，聚合出省报表
    spark.sql(
      """
        |
        |select
        |province,
        |sum(pv_amt)  as pv_amt,
        |card_bm(merge_bitmap(bitmap)) as uv_cnt,
        |merge_bitmap(bitmap) as bitmap
        |
        |from pcr_report
        |group by province
        |
        |""".stripMargin).show(100, false)


    spark.close()

  }

}

SparkSQL项目实战

我们这次 Spark-sql 操作中所有的数据均来自 Hive，首先在 Hive 中创建表,，并导入数据。

一共有3张表： 1张用户行为表，1张城市表，1 张产品表

一共有3张表： 1张用户行为表，1张城市表，1 张产品表
CREATE TABLE `user_visit_action`(
  `date` string,  
  `user_id` bigint,
  `session_id` string,
  `page_id` bigint,
  `action_time` string,
  `search_keyword` string,
  `click_category_id` bigint,
  `click_product_id` bigint,
  `order_category_ids` string,
  `order_product_ids` string,
  `pay_category_ids` string,
  `pay_product_ids` string,
  `city_id` bigint)
row format delimited fields terminated by '\t';
load data local inpath 'input/user_visit_action.txt' into table user_visit_action;

CREATE TABLE `product_info`(
  `product_id` bigint,
  `product_name` string,
  `extend_info` string)
row format delimited fields terminated by '\t';
load data local inpath 'input/product_info.txt' into table product_info;

CREATE TABLE `city_info`(
  `city_id` bigint,
  `city_name` string,
  `area` string)
row format delimited fields terminated by '\t';
load data local inpath 'input/city_info.txt' into table city_info;

需求：各区域热门商品 Top3

需求简介

这里的热门商品是从点击量的维度来看的，计算各个区域前三大热门商品，并备注上每个商品在主要城市中的分布比例，超过两个城市用其他显示。

例如：

地区	商品名称	点击次数	城市备注
华北	商品A	100000	北京21.2%，天津13.2%，其他65.6%
华北	商品P	80200	北京63.0%，太原10%，其他27.0%
华北	商品M	40000	北京63.0%，太原10%，其他27.0%
东北	商品J	92000	大连28%，辽宁17.0%，其他 55.0%

package org.example

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession

object SparkSQL_Req_2 {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setAppName("SparkSql").setMaster("local[*]")
    val spark: SparkSession = SparkSession.builder()
      .config(sparkConf).
      enableHiveSupport().
      getOrCreate()
    val sc: SparkContext = spark.sparkContext
    //准备数据
    spark.sql(
      """
        |select
        |  t3.area,
        |  t3.product_name,
        |  t3.p_click_count,
        |  t3.rk
        |from
        |  (
        |select
        |  t2.area,
        |  t2.product_name,
        |  t2.p_click_count,
        |  rank() over( partition by t2.area order by t2.p_click_count desc ) rk
        |from
        |  (
        |select
        |  t1.area,
        |  t1.product_name,
        |  count(t1.click_product_id) p_click_count
        |
        |from
        |  (
        |select
        |    u.click_product_id ,
        |    p.product_name,
        |    c.city_name,
        |    c.area
        |from
        |   user_visit_action  u
        |join
        |   product_info p
        |on
        |   u.click_product_id = p.product_id
        |join
        |   city_info c
        |on
        |   u.city_id  = c.city_id
        |where
        |   u.click_product_id != -1
        |  )t1
        |group by t1.area , t1.product_name
        |  )t2
        |  )t3
        |where t3.rk <=3
          """.stripMargin).show()

    spark.stop()
    
  }

}

需求分析

查询出来所有的点击记录，并与 city_info 表连接，得到每个城市所在的地区，与 Product_info 表连接得到产品名称

按照地区和商品 id 分组，统计出每个商品在每个地区的总点击次数

每个地区内按照点击次数降序排列

只取前三名

城市备注需要自定义 UDAF 函数

package org.example
import org.apache.spark.SparkConf
import org.apache.spark.sql._
import org.apache.spark.sql.expressions.Aggregator

import scala.collection.mutable
import scala.collection.mutable.ListBuffer
object SparkSQL_Req_3 {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "atguigu")

    //新的起点: SparkSession

    val sparkConf: SparkConf = new SparkConf().setAppName("SparkSQL").setMaster("local[*]")

    val spark: SparkSession =
      SparkSession.builder()
        //.appName("SparkSQL")
        //.master("local[*]")
        .config(sparkConf)
        .enableHiveSupport() // 启用hive的支持
        .getOrCreate()

    var sc = spark.sparkContext
    //写代码不管用不用都导入。

    spark.udf.register("cityMark", functions.udaf(new CityMarkAgg))

    spark.sql(
      """
        |select
        |  t3.area,
        |  t3.product_name,
        |  t3.p_click_count,
        |  t3.c_m,
        |  t3.rk
        |from
        |  (
        |select
        |  t2.area,
        |  t2.product_name,
        |  t2.p_click_count,
        |  t2.c_m ,
        |  rank() over( partition by t2.area order by t2.p_click_count desc ) rk
        |from
        |  (
        |select
        |  t1.area,
        |  t1.product_name,
        |  count(t1.click_product_id) p_click_count,
        |  cityMark(t1.city_name) c_m
        |
        |from
        |  (
        |select
        |    u.click_product_id ,
        |    p.product_name,
        |    c.city_name,
        |    c.area
        |from
        |   user_visit_action  u
        |join
        |   product_info p
        |on
        |   u.click_product_id = p.product_id
        |join
        |   city_info c
        |on
        |   u.city_id  = c.city_id
        |where
        |   u.click_product_id != -1
        |  )t1
        |group by t1.area , t1.product_name
        |  )t2
        |  )t3
        |where t3.rk <=3
         """.stripMargin).show(50, false)

    spark.stop()

  }

  case class CityBuffer(var toatlClick: Long, var cityMap: mutable.Map[String, Long])

  /**
   * 自定义函数(强类型)
   * 1. 继承Aggregator
   * 2. 确定泛型:
   * IN:  String
   * BUF: CityBuffer
   * OUT: String
   */
  class CityMarkAgg extends Aggregator[String, CityBuffer, String] {

    override def zero: CityBuffer = CityBuffer(0L, mutable.Map[String, Long]())

    override def reduce(buffer: CityBuffer, cityName: String): CityBuffer = {
      //总点击次数加1
      buffer.toatlClick += 1

      //城市点击次数加1
      val old: Long = buffer.cityMap.getOrElse(cityName, 0L)
      buffer.cityMap.put(cityName, old + 1)

      buffer
    }

    override def merge(b1: CityBuffer, b2: CityBuffer): CityBuffer = {
      //总点击次数
      b1.toatlClick += b2.toatlClick

      //城市的点击次数
      for ((cityName, cityCount) <- b2.cityMap) {
        val old: Long = b1.cityMap.getOrElse(cityName, 0L)
        b1.cityMap.put(cityName, old + cityCount)
      }
      b1
    }

    override def finish(buffer: CityBuffer): String = {

      val result: ListBuffer[String] = ListBuffer[String]()


      //总点击次数
      val totalClick: Long = buffer.toatlClick
      //城市点击次数
      val citys: List[(String, Long)] = buffer.cityMap.toList.sortBy(_._2)(Ordering.Long.reverse).take(2)

      var totalPer: Double = 100L

      //处理点击比例
      for ((cityName, clickCount) <- citys) {
        val per: Double = clickCount * 100 / totalClick.toDouble
        totalPer -= per
        val cityMark: String = cityName + " " + per + "%"
        result.append(cityMark)
      }

      //处理其他
      if (buffer.cityMap.size > 2) {
        result.append(s"其他 $totalPer%")
      }

      result.mkString(", ")

    }

    override def bufferEncoder: Encoder[CityBuffer] = Encoders.product

    override def outputEncoder: Encoder[String] = Encoders.STRING
  }

}

Spark学习---SparkSQL（概述、编程、数据的加载和保存、自定义UDFA、项目实战）

一、SparkSQL概述

1.1 什么是SparkSQL

1.2 为什么要有SparkSQL

SparkSQL的发展

SparkSQL的特点

二、SparkSQL 编程

2.1 SparkSession 新的起始点

从JSON⽂件加载DataFrame

从非结构化⽂件加载DataFrame

从Parquet⽂件进⾏创建

外部存储服务创建DF

spark整合hive原理--访问hive元数据服务原理

用户自定义函数

自定义UDF实战案例

java开发spark快速上手

自定义UDFA工作逻辑

自定义UDFA实战案例

SparkSQL项目实战

需求：各区域热门商品 Top3

需求简介

需求分析

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

Spark学习---SparkSQL（概述、编程、数据的加载和保存、自定义UDFA、项目实战）

一、SparkSQL概述

1.1 什么是SparkSQL

1.2 为什么要有SparkSQL

SparkSQL的发展

SparkSQL的特点

二、SparkSQL 编程

2.1 SparkSession 新的起始点

从JSON⽂件加载DataFrame

从非结构化⽂件加载DataFrame

从Parquet⽂件进⾏创建

外部存储服务创建DF

spark整合hive原理--访问hive元数据服务原理

用户自定义函数

自定义UDF实战案例

java开发spark快速上手

自定义UDFA工作逻辑

自定义UDFA实战案例

SparkSQL项目实战

需求：各区域热门商品 Top3

需求简介

需求分析

热门文章

最新文章

相关课程

相关电子书

相关实验场景