快捷導(dǎo)航

深入了解SparkSQL的運用及方法

更新時間：2022年03月30日 11:42:31 作者：那人獨釣寒江雪.

SparkSQL就是將SQL轉(zhuǎn)換成一個任務(wù)，提交到集群上運行，類似于Hive的執(zhí)行方式。本文給大家分享了SparkSQl的運用及方法，感興趣的朋友跟隨小編一起看看吧

一：SparkSQL

1.SparkSQL簡介

Spark SQL是Spark的一個模塊，用于處理結(jié)構(gòu)化的數(shù)據(jù)，它提供了一個數(shù)據(jù)抽象DataFrame（最核心的編程抽象就是DataFrame），并且SparkSQL作為分布式SQL查詢引擎。
Spark SQL就是將SQL轉(zhuǎn)換成一個任務(wù)，提交到集群上運行，類似于Hive的執(zhí)行方式。

2.SparkSQL運行原理

將Spark SQL轉(zhuǎn)化為RDD，然后提交到集群執(zhí)行。

3.SparkSQL特點

（1）容易整合，Spark SQL已經(jīng)集成在Spark中

（2）提供了統(tǒng)一的數(shù)據(jù)訪問方式：JSON、CSV、JDBC、Parquet等都是使用統(tǒng)一的方式進行訪問

（3）兼容 Hive

（4）標準的數(shù)據(jù)連接：JDBC、ODBC

二、SparkSQL運用

package sql

import org.apache.avro.ipc.specific.Person
import org.apache.spark
import org.apache.spark.rdd.RDD
import org.apache.spark.sql
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.junit.Test

class Intro {
  @Test
  def dsIntro(): Unit ={
    val spark: SparkSession = new sql.SparkSession.Builder()
      .appName("ds intro")
      .master("local[6]")
      .getOrCreate()

    //導(dǎo)入隱算是shi轉(zhuǎn)換
    import spark.implicits._

    val sourceRDD: RDD[Person] =spark.sparkContext.parallelize(Seq(Person("張三",10),Person("李四",15)))
    val personDS: Dataset[Person] =sourceRDD.toDS();
//personDS.printSchema()打印出錯信息

    val resultDS: Dataset[Person] =personDS.where('age>10)
      .select('name,'age)
      .as[Person]
    resultDS.show()

  }
  @Test
  def dfIntro(): Unit ={
    val spark: SparkSession =new SparkSession.Builder()
      .appName("ds intro")
      .master("local")
      .getOrCreate()

    import spark.implicits._
    val sourceRDD: RDD[Person] = spark.sparkContext.parallelize(Seq(Person("張三",10),Person("李四",15)))
    val df: DataFrame = sourceRDD.toDF()//隱shi轉(zhuǎn)換

    df.createOrReplaceTempView("person")//創(chuàng)建表
    val resultDF: DataFrame =spark.sql("select name from person where age>=10 and age<=20")
    resultDF.show()

  }
  @Test
  def database1(): Unit ={
    //1.創(chuàng)建sparkSession
    val spark: SparkSession =new SparkSession.Builder()
      .appName("database1")
      .master("local[6]")
      .getOrCreate()
      //2.導(dǎo)入引入shi子轉(zhuǎn)換
    import spark.implicits._

    //3.演示
    val sourceRDD: RDD[Person] =spark.sparkContext.parallelize(Seq(Person("張三",10),Person("李四",15)))
    val dataset: Dataset[Person] =sourceRDD.toDS()

    //Dataset 支持強類型的API
    dataset.filter(item => item.age >10).show()
    //Dataset 支持若弱類型的API
    dataset.filter('age>10).show()
    //Dataset 可以直接編寫SQL表達式
    dataset.filter("age>10").show()
  }

  @Test
  def database2(): Unit ={
    val spark: SparkSession = new SparkSession.Builder()
      .master("local[6]")
      .appName("database2")
      .getOrCreate()
    import spark.implicits._

    val dataset: Dataset[Person] =spark.createDataset(Seq(Person("張三",10),Person("李四",20)))
    //無論Dataset中放置的是什么類型的對象,最終執(zhí)行計劃中的RDD上都是internalRow
    //直接獲取到已經(jīng)分析和解析過得Dataset的執(zhí)行計劃,從中拿到RDD
    val executionRdd: RDD[InternalRow] =dataset.queryExecution.toRdd

    //通過將Dataset底層的RDD通過Decoder轉(zhuǎn)成了和Dataset一樣的類型RDD
    val typedRdd:RDD[Person] = dataset.rdd

    println(executionRdd.toDebugString)
    println()
    println()
    println(typedRdd.toDebugString)
  }

  @Test
  def database3(): Unit = {
    //1.創(chuàng)建sparkSession
    val spark: SparkSession = new SparkSession.Builder()
      .appName("database1")
      .master("local[6]")
      .getOrCreate()
    //2.導(dǎo)入引入shi子轉(zhuǎn)換
    import spark.implicits._

    val dataFrame: DataFrame = Seq(Person("zhangsan", 15), Person("lisi", 20)).toDF()
    //3.看看DataFrame可以玩出什么花樣
    //select name from...
    dataFrame.where('age > 10)
      .select('name)
      .show()
  }
//  @Test
//  def database4(): Unit = {
//    //1.創(chuàng)建sparkSession
//    val spark: SparkSession = new SparkSession.Builder()
//      .appName("database1")
//      .master("local[6]")
//      .getOrCreate()
//    //2.導(dǎo)入引入shi子轉(zhuǎn)換
//    import spark.implicits._
//    val personList=Seq(Person("zhangsan",15),Person("lisi",20))
//
//    //1.toDF
//    val df1: DataFrame =personList.toDF()
//    val df2: DataFrame =spark.sparkContext.parallelize(personList).toDF()
//      //2.createDataFrame
//    val df3: DataFrame =spark.createDataFrame(personList)
//
//    //3.read
//    val df4: DataFrame =spark.read.csv("")
//    df4.show()
//  }
  //toDF()是轉(zhuǎn)成DataFrame,toDs是轉(zhuǎn)成Dataset
  //  DataFrame就是Dataset[Row] 代表弱類型的操作,Dataset代表強類型的操作,中的類型永遠是row,DataFrame可以做到運行時類型安全,Dataset可以做到 編譯時和運行時都安全
@Test
def database4(): Unit = {
  //1.創(chuàng)建sparkSession
  val spark: SparkSession = new SparkSession.Builder()
    .appName("database1")
    .master("local[6]")
    .getOrCreate()
  //2.導(dǎo)入引入shi子轉(zhuǎn)換
  import spark.implicits._
  val personList=Seq(Person("zhangsan",15),Person("lisi",20))
  //DataFrame代表弱類型操作是編譯時不安全
  val df: DataFrame =personList.toDF()

  //Dataset是強類型的
  val ds: Dataset[Person] =personList.toDS()
  ds.map((person:Person) =>Person(person.name,person.age))
}
  @Test
  def row(): Unit ={
    //1.Row如何創(chuàng)建,它是什么
    //row對象必須配合Schema對象才會有列名
    val p: Person =Person("zhangsan",15)
    val row: Row =Row("zhangsan",15)
    //2.如何從row中獲取數(shù)據(jù)
    row.getString(0)
    row.getInt(1)
    //3.Row也是樣例類、
    row match {
      case Row(name,age) => println(name,age)
    }
  }

}
case class Person(name: String, age: Int)