檢視 spark-sql 的 SQL語法樹、spark-sql 的優化、整合 hive 之後通過程式碼操作

阿新 • • 發佈：2022-03-13

檢視 spark-sql 的 SQL語法樹、spark-sql 的優化、整合 hive 之後通過程式碼操作

檢視 spark-sql 的 SQL語法樹、spark-sql 的優化、整合 hive 之後通過程式碼操作

一、檢視 spark-sql 的 SQL語法樹

檢視 spark-sql 的web介面的一種方式

在通過 spark-sql --master yarn-client

命令進入 spark-sql 的時候

可以在yarn的web介面中通過下圖所示的步驟進入spark-sql 的web介面，並檢視 spark-sql 的 SQL語法樹

二、spark-sql 的優化

1、快取

新增快取

spark-sql> cache table 表名;

刪除快取
uncache table 表名;

2、廣播小表 -- 實現mapjoin -- hint

在map端實現表關聯，將小表載入到記憶體，小表的大小不能超過一個Executor的記憶體的0.6

//a是小表
spark-sql> 
select /*+broadcast(a)*/ * from 
student as a
join 
score as b
on a.id=b.student_id;

3、廣播小表在程式碼中實現優化

package com.shujia.sql

import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo9MapJoin {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("student")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    //讀取學生表
    val student: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING,name STRING,age INT,gender STRING, clazz STRING")
      .load("data/students.txt")

    //讀取分數表
    val score: DataFrame = spark.read
      .format("csv")
      .option("sep", ",")
      .schema("sId STRING,cId STRING ,sco DOUBLE")
      .load("data/score.txt")


    /**
      * 如果兩表關聯，我們可以手動的將小表廣播出去，提高效率
      * 小表的大小不能超過一個Executor的0.6
      * DF實現mapjoin
      *
      * 將小表廣播實現mapjoin----.hint("broadcast")
      */
	//手動將某個表廣播實現mapjoin
    val joinDF: DataFrame =
      score.join(student.hint("broadcast"), $"sId" === $"id")

    joinDF.show()

    while (true) {

    }
  }
}

三、整合 hive 之後，通過程式碼操作

通過程式碼拿hive資料

在SparkSession要新增.enableHiveSupport()

package com.shujia.sql

import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo10SaprkOnHIve {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("student")
      .config("spark.sql.shuffle.partitions", 1)
      //程式碼在本地不能執行，必須打包到叢集中執行
      .enableHiveSupport() //使用hived的元資料，可以直接在程式碼中使用hive的表
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    //讀取hive中的表
    //student -- hive 中的表
    //通過 spark.table 直接拿
    val student: DataFrame = spark.table("student")
    val score: DataFrame = spark.table("score")

    student
      .join(score, $"id" === $"student_id")
      .show(1000)
  }
}