[Hadoop] Spark SQL - DataFrame & DataSet

我会带着你远行 2022-05-14 05:59 200阅读 0赞

### 1. 基本概念 ###

**分布式计算框架的产生**

*  DataFrame这个词不是Spark SQL独创的，是其他框架已有的（比如说Python的pandas，R语言也有），Spark SQL从其他框架借鉴来的。
 *  Python适用于单机的计算，单机的能处理的数据量很有限。亟待分布式计算框架。
 *  RDD 和MapReduce的学习和开发成本太高，普通程序员开发难度大。

**DataFrame vs DataSet（Spark1.6版本新出来的）**

*  DataSet是一个分布式数据集，DataFrame是有schema的DataSet。DataFrame = DaraSet\[Row\]
 *  DataSet和DataFrame都可以做很多transformation操作。
 *  DataSet API：只支持Java，Scala；不支持Python，R。
 *  DataFrame API：支持Java，Scala，Python，R。

**DataFrame vs RDD**

*  在Spark 1.3版本以前，DataFrame叫SchemaRDD；1.3版本之后，才改叫DataFrame。
 *  RDD没有schema，DataFrame有schema。DataFrame比RDD暴露的信息更多。比如：RDD：Person；DataFrame：Person表，里面有name String，age Int，height Double（可以理解成一张表）。
 *  概念层面：DataFrame和RDD一样，都是一个分布式数据集，都可以并行计算。
 *  API层面：DataFrame各种map filter flatMap...等都支持，且比RDD更为丰富。
 *  数据结构：从底层数据结构看。DataFrame由于有schema详细信息，API使用时性能会更加好。比如：执行Java/Scala API 的过程，底层是 jvm来实现的，也就是说RDD的执行依赖于编程语言自身的运行时环境；而DataFrame ==> Logical Plan (逻辑执行计划)==> Physical Plan，在每个层面都做了优化，不管使用哪一种编程语言(Java/Scala/Python/R)，它都是经过逻辑执行计划到物理执行计划，执行效率是一样的，性能都比RDD更高。

**DataFrame vs RDD vs DataSet（后续更新）**

### **2. Spark SQL编程** ###

Spark SQL入口点：

*  Spark2.0 版本以前：SQLContext    HiveContext
 *  Spark2.0 版本以后：统一入口SparkSession

spark-shell启动时，会启动两个东西：一个sc，一个是spark，这个spark即为spark session。

**构建一个SparkSession**

val spark = SparkSession.builder()    // 这是一个builder设计模式
    .mater(“local[2]”)
    .appName(“SparkSQLApp”)
    .config("spark.some.config.option", "some-value")   // 如果要配置一些参数，就以k/v的形式传进来，不需要配置这句话就不写
    .enableHiveSupport()      // 写了这一句就能访问hive，没有就不能
    .getOrCreate()
    
    
    // 如果需要隐式转换，比如把RDD转成DS则需要
    import spark.implicits._
    
    //使用完要stop
    spark.stop()

**创建一个DF**

// 标准写法：读就用spark.read，写就用df.write
    val df = spark.read.json("/home/hadoop/app/spark/examples/src/main/resources/people.json")
    
    // 源码
    df = spark.read.format("json").load(path)
    df = spark.read.format("text").load(path)
    df = spark.read.format("parquet").load(path)
    df = spark.read.format("orc").load(path)

在spark-shell交互式命令行操作

[hadoop@hadoop000 conf]$ spark-shell --master local[2] --jars ~/software/mysql-connector-java-5.1.46/mysql-connector-java-5.1.46.jar
    
    // 加载json文件把它变成一个df，在下面可以看到df的schema包含两个字段以及字段类型是什么
    scala> val df = spark.read.json("file:///home/hadoop/app/spark/examples/src/main/resources/people.json")
    df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]
    
    scala>

**DF基本使用**

// 查看schema
    scala> df.printSchema
    root
     |-- age: long (nullable = true)   ==> long对应bigint类型
     |-- name: string (nullable = true)
    scala>
    
    //查看DF内容，默认为show(20)
    scala> df.show
    
    //查看指定列
    scala> df.select("name").show
    //或者，这里的name其实是column
    //这种写法在spark-shell交互式命令行可以直接使用是因为spark-shell在启动时已经做了大量的隐式转换。如果我们在IDEA中编程想要用这种写法，要自己import隐式转换进来（import spark.implicits._ ）
    scala> df.select('name).show
    scala> df.select($"name").show
    
    //列计算
    scala> df.select($"name",$"age"+1).show
    
    //过滤
    scala> df.filter($"age">21).show
    
    //分组
    scala> df.groupBy("age").count().show()

**DF Temp/Global视图**

//DF创建临时视图，临时视图是单session的
    scala> df.createOrReplaceTempView("people")
    
    //视图创建以后，就可以使用SQL来操作了，spark.sql的返回值还是DF
    spark.sql("SELECT * FROM people").show()
    
    //Global视图简单了解即可，不常用。Global和Temp视图的生命周期是不一样的。

### 3. DF与RDD进行交互 ###

Spark SQL支持两种方式来转换已经存在的RDDs和DSs：

*  通过反射去推导出RDD的schema信息
 *  通过编程的方式来指定schema信息

**反射推导schema**

//已有infos.txt文件
    [hadoop@hadoop000 data]$ cat /home/hadoop/data/infos.txt
    1,name0,20
    2,name1,18
    3,name2,20
    [hadoop@hadoop000 data]$
    
        //读取文件系统上的文件并创建RDD
        val info = spark.sparkContext.textFile("datas\\infos.txt")
        info.collect().foreach(println)
    
        /**
          * TODO: 将RDD转成DF
          * 因为RDD没有schema，DF有schema，这时候就要用到case class来定义schema，写在下面
          *
          * 具体执行过程：
          * 1. infos.txt文件里面有三行数据，对应转成DF应该也是有三条记录
          * 2. 把RDD中每条数据解析(map)出来：读取每一行的数据，采用逗号分割，分割出来是一个数组。
          * 3. 把数组中每一个对应的元素按照Info类的schema做一下类型转换（因为读进来的元素全部都是string）
          * 4. 最后再转成DF，注意toDF选择不需要带schema类型的
          */
        val infoDF = info.map(_.split(",")).map(x => Info(x(0).toInt, x(1), x(2).toInt)).toDF()
        infoDF.show(false)
        println("=================================")

**编程指定shema**