2021年大数据Spark（二十七）：SparkSQL案例一花式查询和案例二WordCount

package cn.itcast.sql

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * Author itcast
 * Desc 演示SparkSQL的各种花式查询
 */
object FlowerQueryDemo {
  case class Person(id:Int,name:String,age:Int)

  def main(args: Array[String]): Unit = {
    //1.准备环境-SparkSession
    val spark: SparkSession = SparkSession.builder().appName("SparkSQL").master("local[*]").getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("WARN")

    //2.加载数据
    val lines: RDD[String] = sc.textFile("data/input/person.txt")

    //3.切割
    //val value: RDD[String] = lines.flatMap(_.split(" "))//错误的
    val linesArrayRDD: RDD[Array[String]] = lines.map(_.split(" "))

    //4.将每一行(每一个Array)转为样例类(相当于添加了Schema)
    val personRDD: RDD[Person] = linesArrayRDD.map(arr=>Person(arr(0).toInt,arr(1),arr(2).toInt))

    //5.将RDD转为DataFrame(DF)
    //注意:RDD的API中没有toDF方法,需要导入隐式转换!
    import spark.implicits._
    val personDF: DataFrame = personRDD.toDF

    //6.查看约束
    personDF.printSchema()

    //7.查看分布式表中的数据集
    personDF.show(6,false)//false表示不截断列名,也就是列名很长的时候不会用...代替


    //演示SQL风格查询
    //0.注册表名
    //personDF.registerTempTable("t_person")//已经过时
    //personDF.createTempView("t_person")//创建表,如果已存在则报错:TempTableAlreadyExistsException
    //personDF.createOrReplaceGlobalTempView("t_person")//创建全局表,可以夸session使用,查询的时候使用:SELECT * FROM global_temp.表名;生命周期太大,一般不用
    personDF.createOrReplaceTempView("t_person")//创建一个临时表,只有当前session可用!且表如果存在会替换!

    //1.查看name字段的数据
    spark.sql("select name from t_person").show
    //2.查看 name 和age字段数据
    spark.sql("select name,age from t_person").show
    //3.查询所有的name和age，并将age+1
    spark.sql("select name,age,age+1 from t_person").show
    //4.过滤age大于等于25的
    spark.sql("select name,age from t_person where age >=25").show
    //5.统计年龄大于30的人数
    spark.sql("select count(age) from t_person where age >30").show
    //6.按年龄进行分组并统计相同年龄的人数
    spark.sql("select age,count(age) from t_person group by age").show


    //演示DSL风格查询
    //1.查看name字段的数据
    import org.apache.spark.sql.functions._
    personDF.select(personDF.col("name")).show
    personDF.select(personDF("name")).show
    personDF.select(col("name")).show
    personDF.select("name").show

    //2.查看 name 和age字段数据
    personDF.select(personDF.col("name"),personDF.col("age")).show
    personDF.select("name","age").show

    //3.查询所有的name和age，并将age+1
    //personDF.select("name","age","age+1").show//错误,没有age+1这一列
    //personDF.select("name","age","age"+1).show//错误,没有age1这一列
    personDF.select(col("name"),col("age"),col("age")+1).show
    personDF.select($"name",$"age",$"age"+1).show
    //$表示将"age"变为了列对象,先查询再和+1进行计算
    personDF.select('name,'age,'age+1).show
    //'表示将age变为了列对象,先查询再和+1进行计算


    //4.过滤age大于等于25的，使用filter方法/where方法过滤
    personDF.select("name","age").filter("age>=25").show
    personDF.select("name","age").where("age>=25").show

    //5.统计年龄大于30的人数
    personDF.where("age>30").count()

    //6.按年龄进行分组并统计相同年龄的人数
    personDF.groupBy("age").count().show

  }

}

package cn.itcast.sql

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

/**
 * Author itcast
 * Desc 使用SparkSQL完成WordCount---SQL风格和DSL风格
 */
object WordCount {
  def main(args: Array[String]): Unit = {
    //1.准备环境
    val spark: SparkSession = SparkSession.builder().appName("WordCount").master("local[*]").getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("WARN")
    import spark.implicits._

    //2.加载数据
    //val rdd: RDD[String] = sc.textFile("data/input/words.txt")//可以使用该方式,然后使用昨天的知识将rdd转为df/ds
    val df: DataFrame = spark.read.text("data/input/words.txt")
    val ds: Dataset[String] = spark.read.textFile("data/input/words.txt")
    //df.show()//查看分布式表数据
    //ds.show()//查看分布式表数据
    

    //3.做WordCount
    //切割
    //df.flatMap(_.split(" ")) //注意:直接这样写报错!因为df没有泛型,不知道_是String!
    //df.flatMap(row=>row.getAs[String]("value").split(" "))
    val wordsDS: Dataset[String] = ds.flatMap(_.split(" "))
    //wordsDS.show()
    
    //使用SQL风格做WordCount
    wordsDS.createOrReplaceTempView("t_words")
    val sql:String =
      """
        |select value,count(*) as count
        |from t_words
        |group by value
        |order by count desc
        |""".stripMargin
    spark.sql(sql).show()

    //使用DSL风格做WordCount
    wordsDS
      .groupBy("value")
      .count()
      .orderBy($"count".desc)
      .show()
    
    /*
    +-----+-----+
    |value|count|
    +-----+-----+
    |hello|    4|
    |  her|    3|
    |  you|    2|
    |   me|    1|
    +-----+-----+
    
    +-----+-----+
    |value|count|
    +-----+-----+
    |hello|    4|
    |  her|    3|
    |  you|    2|
    |   me|    1|
    +-----+-----+
     */

  }
}