当前位置:   article > 正文

Spark-SQL——DataFrame与Dataset_java datasets spark.sql

java datasets spark.sql

一、Spark SQL概述

1.1、Spark SQL是什么?

Spark SQL是Apache Spark中用于结构化数据处理的模块。它允许开发人员在Spark上执行SQL查询、处理结构化数据以及将它们与常规的RDD一起使用。Spark Sql提供了用于处理结构化数据的高级API,如DataFrames和Datasets,它们比原始的RDD API更加高效和方便。
通过Spark SQL,可以使用标准的SQL语言进行数据处理,也可以使用DataFrame API进行操作。此外,Spark SQL还可以集成Hive,允许用户直接查询Hive表并将查询结果存储到Hive中。Spark SQL也支持Spark中运行标准的JDBC/ODBC连接,从而可以使用各种外部数据源。

1.2、Hive和Spark SQL

Hive和Spark SQL都是用来处理大数据的工具,主要是基于Hadoop生态圈。它们的相同点都是用来查询和处理大规模数据的,而且都可以使用类SQL语言来进行操作。

不同之处:

  • 操作语言不同:Hive使用HQL(Hive Query Language)进行数据操作,而Spark SQL使用Spark SQL语法进行数据操作。
  • 数据处理方式不同:Hive依赖于MapReduce作为计算引擎,而Spark SQL使用Spark作为计算引擎。由于Spark的内存计算能力较强,所以在某些场景下,Spark SQL比Hive更加高效。
  • 数据存储方式不同:Hive使用HDFS或者其他支持Hadoop HDFS API的存储系统来存储数据,而Spark SQL可以支持多种不同的数据存储系统,例如:HDFS、Hive等。
  • 性能不同:Spark SQL的性能要比Hive快得多,主要是因为Spark Sql使用了内存计算技术,而Hive使用的是MapReduce计算模型。
  • 执行计划不同:Hive的执行计划是通过HQL生成的,而Spark Sql的执行计划是通过Spark的优化器生成的。Spark的优化器可以对查询进行优化,以提高查询的性能。

1.3、DataFrame与DataSet

  • DataFrame
    • DataFrame是Spark SQL的一种数据抽象,它表示分布式数据集合。DataFrame和关系型数据库中的表类似,都有列和行的概念,而且还具备了分布式的特性。DataFrame提供了丰富的数据操作接口,例如:选择、过滤、分组、聚合、排序、连接等。这些操作和SQL语句类似,便于使用和理解。
    • DataFrame可以看作是由一系列Row对象组成的RDD,每个Row代表着一行数据。不同于RDD,DataFrame具有数据类型的概念,这意味着DataFrame的每一列都有其对应的数据类型,这一特性使得Spark能够进行更为智能的优化,例如通过列式存储来提升查询效率。同时DataFrame的API使用了强类型语言的特性,使得在编译期就能够捕获到许多错误,提高了开发效率。
  • Dataset
    • Dataset是Spark中的一种强类型API,它结合了Spark SQL中的DataFrame和RDD的优点。
    • Dataset API提供了编译时类型安全和面向对象的编程接口,并利用Spark SQL的优化执行引擎来提高查询性能。它可以使用Scala、Java和Python等语言进行操作,支持多种数据源,如Parquet、JSON、JDBC等。
    • Dataset API还提供了对关系型数据和面向对象数据的支持,可将其转换为Dataset,通过统一的编程接口进行处理和操作。
  • 二者不同:
    • 数据类型:DataFrame 是一种以列为基础的分布式数据集合,强调数据的结构,即 schema,可以使用 SQL 或者 DataFrame 的 API 进行操作;Dataset 是一种类型化的分布式数据集合,强调类型的安全性,可以使用强类型的 API 进行操作。
    • 类型检查:DataFrame 只有在运行时才会进行类型检查,可能在运行时出现类型错误;Dataset 利用 Spark 强大的编译器来捕捉类型错误,可以在编译时就发现类型错误。
    • API:DataFrame 的 API 是基于 DataFrame 和 SQL 的 API,使用较为灵活,但是对于类型的检查较为弱;Dataset 的 API 是基于 Scala 和 Java 的编程语言,可以保证类型的安全性,但是使用的时候会相对复杂一些。
    • 编译时优化:Dataset 可以受益于 Spark 的 Catalyst 查询优化器,在编译时进行优化,可以提高性能;DataFrame 只有在运行时才能够获得优化。
    • 语言支持:DataFrame 可以使用 Scala,Java,Python 和 R 等编程语言,比 Dataset 支持的编程语言更多。Dataset 只支持 Scala 和 Java。
    • 性能:由于 Dataset 强制类型检查,所以可以获得更好的性能,因此在需要高性能的场景下推荐使用 Dataset。但是对于某些特定场景下,DataFrame 可能会更适合。

二、Spark SQL编程

Spark Core 中,如果想要执行应用程序,需要首先构建上下文环境对象 SparkContext,Spark SQL 其实可以理解为对 Spark Core 的一种封装,不仅仅在模型上进行了封装,上下文环境对象也进行了封装。
SparkSession 是 Spark 最新的 SQL 查询起始点,实质上是 SQLContext 和 HiveContext的组合,所以在 SQLContext 和 HiveContext 上可用的 API 在 SparkSession 上同样是可以使用的。SparkSession 内部封装了 SparkContext,所以计算实际上是由 sparkContext 完成的。当我们使用 spark-shell 的时候, spark 框架会自动的创建一个名称叫做 spark 的 SparkSession 对象, 就像我们以前可以自动获取到一个 sc 来表示 SparkContext 对象一样
在这里插入图片描述

2.1、DataFrame

2.1.1、创建DataFrame

在Spark SQL中SparkSession是创建DataFrame和执行SQL的入口,创建DataFrame有三种方式:通过Spark的数据源进行创建;从一个存在的RDD进行转换;还可以从HiveTable进行查询返回。

  • 从Spark数据源进行创建
    • Spark支持的创建文件的数据源格式:
      在这里插入图片描述
    • 创建文件user.json文件
      {“username”:“zhangsan”,“age”:20}
    • 读取json文件创建DataFrame
      scala> spark.read.json("file:///opt/stufile/user.json")
      res2: org.apache.spark.sql.DataFrame = [age: bigint, username: string]          
      
      • 1
      • 2
    • 结果
      scala> res2.show()
      +---+--------+
      |age|username|
      +---+--------+
      | 20|zhangsan|
      +---+--------+
      
      • 1
      • 2
      • 3
      • 4
      • 5
      • 6
  • RDD与DataFrame互转
    • 在IDEA中开发程序时,如果需要RDD与DF或者DS之间进行互相操作,那么需要引入 import spark.implicits._
    • 在spark-shell中无需导入,自动完成此操作
    • 创建样例类
      scala> case class User(name:String,age:Int)
      defined class User
      
      • 1
      • 2
    • 创建RDD
      sc.makeRDD(List(("zhangsan",30),("lisi",20)))
      res4: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[8] at makeRDD at <console>:25
      
      
      • 1
      • 2
      • 3
    • 转换DataFrame
      scala> res5.map(t => User(t._1,t._2)).toDF
      res7: org.apache.spark.sql.DataFrame = [name: string, age: int]
      
      scala> res7.show
      +--------+---+
      |    name|age|
      +--------+---+
      |zhangsan| 30|
      |    lisi| 20|
      +--------+---+
      
      
      • 1
      • 2
      • 3
      • 4
      • 5
      • 6
      • 7
      • 8
      • 9
      • 10
      • 11
    • 转换为RDD
      scala> res7.rdd
      res9: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[21] at rdd at <console>:26
      
      
      • 1
      • 2
      • 3

2.1.2、SQL语法

SQL语法风格是指我们查询数据的时候使用SQL语句来查询,这种风格的查询必须要有临时视图或者全局视图来辅助。

df.createOrReplaceTempView(“people”) 创建临时表(创建或重写临时表)
df.createOrReplaceGlobalTempView(“people”) 创建全局表(创建或重写全局表)
**注意:**普通临时表是Session范围内的,如果想要应用范围内有效,可以使用全局临时表。使用全局临时表是需要全路径访问,前面加上global_temp. 例如:select * from global_temp.people

  • 读取JSON文件创建DataFrame
scala> val df = spark.read.json("file:///opt/stufile/user.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, username: string] 
  • 1
  • 2
  • 对DataFrame创建一个临时表
scala> df.createOrReplaceTempView("people")
  • 1
  • 通过SQL语句实现查询全表
scala> val sqlDF = spark.sql("SELECT * FROM people")
sqlDF: org.apache.spark.sql.DataFrame = [age: bigint, name: string]
  • 1
  • 2
  • 结果展示
scala> sqlDF.show
+---+--------+
|age|username|
+---+--------+
| 20|zhangsan|
+---+--------+
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 创建全局表
scala> df.createOrReplaceGlobalTempView("people")
  • 1
  • 查询全表
scala> spark.sql("SELECT * FROM global_temp.people").show()
+---+--------+
|age|username|
+---+--------+
| 20|zhangsan|
+---+--------+
scala> spark.newSession().sql("SELECT * FROM global_temp.people").show()
+---+--------+
|age|username|
+---+--------+
| 20|zhangsan|
+---+--------+
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12

2.1.3、DSL语法

DataFrame提供一个特定领域语言(domain-specific language,DSL)去管理结构化的数据。可以在Scala、Java、Python等语言中使用DSL,使用DSL语法风格不必去创建临时表。

  • 创建一个DataFrame
    scala> val df = spark.read.json("file:///opt/stufile/user.json")
    df: org.apache.spark.sql.DataFrame = [age: bigint, username: string] 
       
    
    • 1
    • 2
    • 3
  • 查看DataFrame的Schema信息
    	scala> df.printSchema
    	root
    	 |-- age: long (nullable = true)
    	 |-- username: string (nullable = true)
    
    • 1
    • 2
    • 3
    • 4
  • 只查看username列数据
    scala> df.select("username").show()
    +--------+
    |username|
    +--------+
    |zhangsan|
    | lisi|
    | wangwu|
    +--------+
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
  • 查看username列数据以及age+1数据
    scala> df.select($"username",$"age" + 1).show
    scala> df.select('username, 'age + 1).show()
    scala> df.select('username, 'age + 1 as "newage").show()
    +--------+---------+
    |username|(age + 1)|
    +--------+---------+
    |zhangsan| 21|
    | lisi| 31|
    | wangwu| 41|
    +--------+---------+
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
  • 查看age大于30的数据
    scala> df.filter($"age">30).show
    +---+---------+
    |age| username|
    +---+---------+
    | 40| wangwu|
    +---+---------+
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
  • 按照age分组,查看数据条数
    scala> df.groupBy("age").count.show
    +---+-----+
    |age|count|
    +---+-----+
    | 20| 1|
    | 30| 1|
    | 40| 1|
    +---+-----+
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

2.2、Dataset

DataSet是具有强类型的数据集合,需要提供对于的类型信息。

2.2.1、创建DataSet

  • 使用样例类序列创建Dataset
    scala> case class Person(name: String, age: Long)
    defined class Person
    scala> val caseClassDS = Seq(Person("zhangsan",2)).toDS()
    caseClassDS: org.apache.spark.sql.Dataset[Person] = [name: string, age: Long]
    scala> caseClassDS.show
    +---------+---+
    | name|age|
    +---------+---+
    | zhangsan| 2|
    +---------+---+
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
  • 使用基本类型的序列创建Dataset
    scala> val ds = Seq(1,2,3,4,5).toDS
    ds: org.apache.spark.sql.Dataset[Int] = [value: int]
    scala> ds.show
    +-----+
    |value|
    +-----+
    | 1|
    | 2|
    | 3|
    | 4|
    | 5|
    +-----+
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12

2.2.2、RDD与Dataset互转

  • RDD转换Dataset

    SparkSQL 能够自动将包含有 case 类的 RDD 转换成 DataSet,case 类定义了 table 的结构,case 类属性通过反射变成了表的列名。Case 类可以包含诸如 Seq 或者 Array 等复杂的结构。

    scala> case class User(name:String, age:Int)
    defined class User
    
    scala> sc.makeRDD(List(("zhangsan",30), ("lisi",49))).map(t=>User(t._1, t._2)).toDS
    res0: org.apache.spark.sql.Dataset[User] = [name: string, age: int]
    
    scala> res0.show
    +--------+---+                                                                  
    |    name|age|
    +--------+---+
    |zhangsan| 30|
    |    lisi| 49|
    +--------+---+
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
  • Dataset转换为RDD

    scala> case class User(name:String, age:Int)
    defined class User
    scala> sc.makeRDD(List(("zhangsan",30), ("lisi",49))).map(t=>User(t._1,t._2)).toDS
    res11: org.apache.spark.sql.Dataset[User] = [name: string, age: int]
    scala> val rdd = res11.rdd
    rdd: org.apache.spark.rdd.RDD[User] = MapPartitionsRDD[51] at rdd at <console>:25
    scala> rdd.collect
    res12: Array[User] = Array(User(zhangsan,30), User(lisi,49))
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

2.2.3、DataFrame 和 DataSet 转换

  • DataFrame 转换为 DataSet
    scala> case class User(name:String, age:Int)
    defined class User
    scala> val df = sc.makeRDD(List(("zhangsan",30),
    ("lisi",49))).toDF("name","age")
    df: org.apache.spark.sql.DataFrame = [name: string, age: int]
    scala> val ds = df.as[User]
    ds: org.apache.spark.sql.Dataset[User] = [name: string, age: int]
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
  • DataSet 转换为 DataFrame
    scala> val ds = df.as[User]
    ds: org.apache.spark.sql.Dataset[User] = [name: string, age: int]
    scala> val df = ds.toDF
    df: org.apache.spark.sql.DataFrame = [name: string, age: int]
    
    • 1
    • 2
    • 3
    • 4

在这里插入图片描述

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/你好赵伟/article/detail/622573
推荐阅读
相关标签
  

闽ICP备14008679号