• JavaSpark | SparkSQL | 创建DataSet | UDF与UDAF | 开窗函数

    文章目录

    一、SparkSQL

    1.SparkSQL介绍

      Hive 是 Shark 的前身，Shark 是 SparkSQL 的前身，SparkSQL 产生的根本原因是其完全脱离了 Hive 的限制。

    • SparkSQL 支持查询原生的 RDD。RDD 是 Spark 平台的核心概念， 是 Spark 能够高效的处理大数据的各种场景的基础。
    • 能够在 scala 、Java中写 SQL 语句。支持简单的 SQL 语法检查，能够在Scala 中写 Hive 语句访问 Hive 数据，并将结果取回作为 RDD 使用。
    • Spark on Hive： Hive 只作为储存角色，Spark 负责 sql 解析优化，执行。
    • Hive on Spark：Hive 即作为存储又负责 sql 的解析优化，Spark 负责执行。

    两者数据源均为Hive表，底层人物均为Spark人物，关键区别在于一个是Hive去解析，一个是Spark去解析。

    2.Dataset 与 DataFrame概念解析

    在这里插入图片描述

      Dataset 也是一个分布式数据容器。与 RDD 类似，然而 Dataset 更像传统数据库的二维表格，除了数据以外，还掌握数据的结构信息（元数据），即schema。同时，与 Hive 类似，Dataset 也支持嵌套数据类型（struct、array 和 map）。从 API 易用性的角度上 看， Dataset API 提供的是一套高层的关系操作，比函数式的 RDD API 要更加友好，门槛更低。Dataset 的底层封装的是 RDD，当 RDD 的泛型是 Row 类型的时候，我们也可以称它为 DataFrame。即 Dataset = DataFrame

    3.SparkSQL 的数据源

      SparkSQL 的数据源可以是 JSON 类型的字符串，JDBC，Parquent，Hive，HDFS 等。
    在这里插入图片描述

    4.SparkSQL 底层架构

      首先拿到 sql 后解析一批未被解决的逻辑计划，再经过分析得到分析后的逻辑计划，再经过一批优化规则转换成一批最佳优化的逻辑计划，再经过 SparkPlanner 的策略转化成一批物理计划，随后经过消费模型转换成一个个的 Spark 任务执行。
    在这里插入图片描述

    5.谓词下推（predicate Pushdown）

    在这里插入图片描述

    二、创建DataSet的几种方式

    1.读取 json 格式的文件创建 Dataset

    • json 文件中的 json 数据不能嵌套 json 格式数据。
    • Dataset 是一个一个 Row 类型的 RDD，ds.rdd()/ds.javaRdd()。
    • 可以两种方式读取 json 格式的文件。
    • df.show()默认显示前 20 行数据。
    • Dataset 原生 API 可以操作 Dataset（不方便）。
    • 注册成临时表时，表中的列默认按 ascii 顺序显示列。
    package com.shsxt.java_Test.sql.dataset;


import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

public class CreateDSFromJosonFile {
   

    public static void main(String[] args) {
   

        SparkSession sparkSession = SparkSession
                .builder()
                .appName("jsonfile")
                .master("local")
                .getOrCreate();


        /**
         * Dataset的底层是一个一个的RDD
         * 注意：
         *      当 Dataset里的泛型是Row时，我们又可以称之为dataframe
         *
         * 以下两种方式都可以读取json格式的文件
         *
         */

        Dataset<Row> ds = sparkSession.read().format("json").load("data/json");
//		Dataset ds = sparkSession.read().json("data/json");
        ds.show();


        /**
         * 显示DataSet中的内容，默认显示前20行。如果现实多行要指定多少行show(行数)
         * 注意：当有多个列时，显示的列先后顺序是按列的ascii码先后显示。
         */
        ds.show(100);


        /**
         *DataSet转换成RDD
         */
//        JavaRDD javaRDD = ds.javaRDD();

        /**
         * 树形的形式显示schema信息
         */
        ds.printSchema();


        /**
         * dataset自带的API 操作dataset
         */
		//select name from table
		ds.select("name").show();
		//select name ,age+10 as addage from table
		ds.select(ds.col("name"),ds.col("age").plus(10).alias("addage")).show();
		//select name ,age from table where age>19
		ds.select(ds.col("name"),ds.col("age")).where(ds.col("age").gt(19)).show();
		//select age,count(*) from table group by age
		ds.groupBy(ds.col("age")).count().show();

        /**
         * 将dataset注册成临时的一张表，这张表相当于临时注册到内存中，逻是辑上的表，不会雾化到磁盘
         */

        ds.createOrReplaceTempView("jtable");
        //ds.registerTempTable("jtable");

        Dataset<Row> sql = sparkSession.sql("select age,count(*) as gg from jtable group by age");
        sql.show();
//
//        Dataset sql2 = sparkSession.sql("select name,age from jtable");
//        sql2.show();
        sparkSession.stop();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    +----+--------+
| age|    name|
+----+--------+
|  18|zhangsan|
|null|    lisi|
|  18|  wangwu|
|  28|  laoliu|
|  20|zhangsan|
|null|    lisi|
|  18|  wangwu|
|  28|  laoliu|
|  28|zhangsan|
|null|    lisi|
|  18|  wangwu|
+----+--------+

+----+--------+
| age|    name|
+----+--------+
|  18|zhangsan|
|null|    lisi|
|  18|  wangwu|
|  28|  laoliu|
|  20|zhangsan|
|null|    lisi|
|  18|  wangwu|
|  28|  laoliu|
|  28|zhangsan|
|null|    lisi|
|  18|  wangwu|
+----+--------+

root
 |-- age: long (nullable = true)
 |-- name: string (nullable = true)

+--------+
|    name|
+--------+
|zhangsan|
|    lisi|
|  wangwu|
|  laoliu|
|zhangsan|
|    lisi|
|  wangwu|
|  laoliu|
|zhangsan|
|    lisi|
|  wangwu|
+--------+

+--------+------+
|    name|addage|
+--------+------+
|zhangsan|    28|
|    lisi|  null|
|  wangwu|    28|
|  laoliu|    38|
|zhangsan|    30|
|    lisi|  null|
|  wangwu|    28|
|  laoliu|    38|
|zhangsan|    38|
|    lisi|  null|
|  wangwu|    28|
+--------+------+

+--------+---+
|    name|age|
+--------+---+
|  laoliu| 28|
|zhangsan| 20|
|  laoliu| 28|
|zhangsan| 28|
+--------+---+

+----+---+
| age| gg|
+----+---+
|null|  3|
|  28|  3|
|  18|  4|
|  20|  1|
+----+---+

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85

    2.通过 json 格式的 RDD 创建 Dataset

    package com.shsxt.java_Test.sql.dataset;


import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.*;


import java.util.Arrays;

/**
 * 读取json格式的RDD创建DF
 * @author root
 *
 */
public class CreateDSFromJsonRDD {
   
	public static void main(String[] args) {
   
        SparkSession sparkSession = SparkSession
                .builder()
                .appName("jsonrdd")
                .master("local")
                .getOrCreate();

        /**
         * 注意：
         *   1.由于是java版,故通过javaSparkcontext.parallelize来创建json格式的JavaRDD
         *   所以我们通过sparkcontext来创建javaSparkcontext
         *   2.如果是scala版本，直接通过sparkcontext.parallelize来创建，就无需创建javaSparkcontext
         */

        SparkContext sc = sparkSession.sparkContext();

        JavaSparkContext jsc = new JavaSparkContext(sc);

        JavaRDD<String> nameRDD = jsc.parallelize(Arrays.asList(
					"{'name':'zhangsan','age':\"18\"}",
					"{\"name\":\"lisi\",\"age\":\"19\"}",
					"{\"name\":\"wangwu\",\"age\":\"20\"}"
				));

		JavaRDD<String> scoreRDD = jsc.parallelize(Arrays.asList(
				"{\"name\":\"zhangsan\",\"score\":\"100\"}",
				"{\"name\":\"lisi\",\"score\":\"200\"}",
				"{\"name\":\"wangwu\",\"score\":\"300\"}"
				));

		Dataset<Row> nameds = sparkSession.read().json(nameRDD);
        Dataset<Row> scoreds = sparkSession.read().json(scoreRDD);


		//注册成临时表使用
        nameds.createOrReplaceTempView("nameTable");
        scoreds.createOrReplaceTempView("scoreTable");

        Dataset<Row> result =
                sparkSession.sql("select nameTable.name,nameTable.age,scoreTable.score "
							+ "from nameTable join scoreTable "
							+ "on nameTable.name = scoreTable.name");
		result.show();
		sc.stop();
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    +--------+---+-----+
|    name|age|score|
+--------&#
    • 1
    • 2
  • 相关阅读:
    JVM之方法区
    02-MySQL库和表的操作
    HCIA --- VLAN实验配置
    持续集成和持续部署（CI/CD）
    C#控制电脑注销、关机、重启
    面经-虚拟机-对象引用类型
    Office Word如何设置页码（MacOS中为例）
    如何测量晶振的频率
    mybatis实现批量更新修改(性能极佳)
    【附源码】计算机毕业设计JAVA校园讲座管理
  • 原文地址：https://blog.csdn.net/liujiesxs/article/details/125970515