SparkSQL 简单总结一下和一个WC案例

分手后的思念是犯贱 2023-06-15 07:11 2阅读 0赞

## SparkSQL总结： ##

### 一、Spark SQL ###

是Spark用来处理结构化数据的一个模块,它提供了一个编程抽象叫做DataFrame,并且作为分布式SQL查询引擎的作用

### 二、什么是DataFrames?? ###

与RDD类似,DataFrames也是一个分布式数据容器;然而DataFrame更像传统数据库的二维表格,除了数据以外,还记录数据的结构信息,即schema;  
同时与Hive类似,DataFrame也支持嵌套数据类型(struct,array,map).

### 三、创建dataframe的两种方式： ###

通过数据源采集直接创建成DF格式(spark.read.json(Path))  
通过StructType创建Schema，通过createDataFrame来创建DF  
DataFrame与DataSet与RDD互相转换  
（1）DataFrame -> DataSet  
val dataSet = dataFrame.as\[Person\]  
（2）DataSet -> DataFrame  
val dataFraem = dataSet.toDF()  
（3）DataFrame -> RDD  
val RDD = dataFrame.rdd

### 四、DataFrame两种风格语法常用操作 ###

1、DSL（领域特定语言）风格操作  
略  
2、SQL风格操作  
DataFrame的一个强大之处就是我们可以将它看作是一个关系型数据表，然后可以通过在程序中使用spark.sql() 来执行SQL语句查询，结果返回一个DataFrame:  
spark.sql(“SQL语句”).show  
如果想使用SQL风格的语法，需要将DataFrame注册成表：  
personDF.registerTempTable(“t\_person”)

### 五、DataSet和DataFrame的区别 ###

1、DataSet包含了DataFrame的功能，Spark2.0中两者统一，DataFrame表示为DataSet\[Row\]，即DataSet的子集。  
2、相比DataFrame，Dataset提供了编译时类型检查，对于分布式程序来讲，提交一次作业太费劲了（要编译、打包、上传、运行），  
到提交到集群运行时才发现错误，这会浪费大量的时间，这也是引入Dataset的一个重要原因。

3、DS状态的RDD\[Person\]虽然以Person为类型参数，但Spark框架本身不了解 Person类的内部结构。  
而DataFrame却提供了详细的结构信息，使得Spark SQL可以清楚地知道该数据集中包含哪些列，每列的名称和类型各是什么，DataFrame多了数据的结构信息，  
即schema。这样看起来就像一张表了。  
4、DataFrame通过引入schema和off-heap（不在堆里面的内存，指的是除了不在堆的内存，使用操作系统上的内存），解决了RDD的缺点,  
Spark通过schame就能够读懂数据, 因此在通信和IO时就只需要序列化和反序列化数据, 而结构的部分就可以省略了；通过off-heap引入，  
可以快速的操作数据，避免大量的GC。但是却丢了RDD的优点，DataFrame不是类型安全的, API也不是面向对象风格的

5、off-heap  
DataFrame还引入了off-heap,意味着JVM堆以外的内存, 这些内存直接受操作系统管理（而不是JVM）。Spark能够以二进制的形式序列化数据(不包  
括结构)到off-heap中, 当要操作数据时, 就直接操作off-heap内存. 由于Spark理解schema, 所以知道该如何操作。

### 六、SparkSQL实现的WordCount ###

import org.apache.spark.sql.{
        Dataset, SparkSession}
    
    /**
     * @description ${
        DESCRIPTION}
     * @author YDAlex
     * @data 2019/11/20.
     * @version 1.0
     */
    object wc {
        
    
      def main(args: Array[String]): Unit = {
        
        val spark = SparkSession.builder()
          .appName("wordCount")
          .master("local[*]")
          .getOrCreate()
        val words = spark.read
          .textFile("G:\\IntelliJ IDEA 2019.2.1\\Workspace\\sparkStreaming\\data\\wc.txt")
        import spark.implicits._
        val word: Dataset[String] = words.flatMap(_.split(" "))
        //SQL style
        word.createOrReplaceTempView("word")//创建临时表
        /*SELECT value, COUNT(*) counts FROM word GROUP BY value ORDER BY counts DESC*/
        val result = spark.sql(
          """
            |select
            |value,
            |count(*) counts
            |from word
            |group by value
            |order by counts desc
            |""".stripMargin)
        result.show()
    
        //DSLStyle
        import org.apache.spark.sql.functions._
        word.groupBy($"value" as "word")
          .agg(count("*") as "counts")
          .orderBy($"counts" desc)
          .show()
    
    
        spark.close()
      }
    }