（10）spark sql 笔记--01基本使用

数据结构上下文环境对象Dataframe

语法风格 DataSetRDD & Dataframe & DataSet 三者 相互转化UDF函数

有两种数据结构：

Dataframe: 在原有RDD的基础上 加上一个结构（类似于字段名），如下
（数据类型是弱类型，靠spark自动推断）

+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+

DataSet:是具有强类型的数据集合，需要提供对应的类型信息。
一般使用样例类包装
（在spark1.6之后，在Dataframe的基础上，增加字段类型。类似于一张hive表同时具有了字段名和字段类型。）

SparkCore中，要执行应用程序，首先得创建SparkContext上下文环境对象
那么在spark sql 中，环境对象为SparkSession ，内部封装了SparkContext

SQL风格 && DSL风格

数据准备 student.json

{"name": "zhangsan","age": 15}
{"name": "lisi","age": 15}
{"name": "wangwu","age": 15}

sparksql代码

  def main(args: Array[String]): Unit = {
    //创建sparkSession环境对象
    val sparkSession = SparkSession.builder()
      .appName("sql")
      .master("local[*]")
      .getOrCreate()
    //导入隐式转换
	import sparkSession.implicits._
   
    //创建Dataframe
    val df: Dataframe = sparkSession.read.json("datas/student.json")
    
    println("****************SQL风格******************************")
   //TODO SQL风格 需要先创建临时视图，再使用sql查询
    df.createOrReplaceTempView("student")
    sparkSession.sql("select * from student").show()
    //普通临时表是Session范围内的，如果想全局应用范围内有效，可以使用全局临时表
    //df.createGlobalTempView("people")
    //sparkSession.sql("SELECt * FROM global_temp.people").show()

    println("****************DSL风格******************************")
    //TODO DSL风格 直接调用方法查询
    df.select($"age" + 1).show()
    
  }

结果展示

****************SQL风格******************************
+---+--------+
|age|    name|
+---+--------+
| 15|zhangsan|
| 15|    lisi|
| 15|  wangwu|
+---+--------+

****************DSL风格******************************
+---+
|age|
+---+
| 16|
| 16|
| 16|
+---+

//定义具有 字段名 和字段类型 得样例类
case class Person(name: String, age: Long)
//创建DataSet
val caseClassDS = Seq(Person("Andy", 32)).toDS()

object sparkSql {
  def main(args: Array[String]): Unit = {
    //创建spark环境对象
    val spark = SparkSession.builder()
      .appName("sql")
      .master("local[*]")
      .getOrCreate()
    
    import spark.implicits._
    
   //RDD <=> Dataframe
    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("lisi",10),("jesh",20),("helly",40),("tom",27)))
    val dataframe: Dataframe = rdd.toDF("name","age")
    
    val newRdd: RDD[Row] = dataframe.rdd

    //Dataframe <=> DataSet
    val dataSet: Dataset[People] = dataframe.as[People]
    val newDf: Dataframe = dataSet.toDF()

    //RDD <=> DataSet
    val ds: Dataset[People] = rdd.map{
      case (name,age)=>{
        People(name,age)
      }
    }.toDS()

    val rdd1: RDD[People] = ds.rdd
  }
  case class People(name:String,age:Int)
}

用户自定义函数

  def main(args: Array[String]): Unit = {
    //创建spark环境对象
    val spark = SparkSession.builder()
      .appName("sql")
      .master("local[*]")
      .getOrCreate()

    //创建Dataframe
    val df: Dataframe = spark.read.json("datas/student.json")
    //TODO SQL风格 需要先创建临时视图，再使用sql查询
    df.createOrReplaceTempView("student")

    //注册用户自定义函数
    spark.udf.register("prefixName",(name:String) =>{
      "Name: " + name
    })

    //使用udf函数给name加前缀
    spark.sql("select prefixName(name) from student").show()
  }

打印结果：

+--------------------+
|UDF:prefixName(name)|
+--------------------+
|      Name: zhangsan|
|          Name: lisi|
|        Name: wangwu|
+--------------------+