Spark SQL快速入門系列之Hive

點選上方藍色字型，選擇“設為星标”

回複”資源“擷取更多資源

大資料技術與架構

點選右側關注，大資料開發領域最強公衆号！

暴走大資料

點選右側關注，暴走大資料！

目錄

• 一.hive和spark sql的內建方式(面試可能會問到)
• 二.spark_shell和spark_sql操作
• • spark_shell
  • spark_sql
  • 使用hiveserver2 + beeline
• 三.腳本使用spark-sql
• 四.idea中讀寫Hive資料
• • 1.從hive中讀資料
  • 2.從hive中寫資料
  • 使用hive的insert語句去寫
  • 使用df.write.saveAsTable("表名")(常用)
  • 使用df.write.insertInto("表名")
  • 3.saveAsTable和insertInto的原理
• 五.聚合後的分區數

hive on spark(版本相容)

官網https://cwiki.apache.org/confluence/display/Hive/Hive+on+Spark%3A+Getting+Started

spark on hive(版本相容)

官網

http://spark.apache.org/docs/2.1.1/sql-programming-guide.html#hive-tables

如果你在叢集上使用了tez，你需要在spark/conf下spark-defaults.conf添加lzo的路徑

spark.jars=/export/servers/hadoop-2.7.7/share/hadoop/common/hadoop-lzo-0.4.20.jar
           

spark-yarn模式啟動

bin/spark-shell --master yarn
           

完全跟sql一樣

spark-sql 得到的結果不夠友好, 是以可以使用hiveserver2 + beeline

1.啟動thriftserver(背景)

sbin/start-thriftserver.sh
           

2.啟動beeline

bin/beeline
# 然後輸入
!connect jdbc:hive2://hadoop102:10000
# 然後按照提示輸入使用者名和密碼
           

添加依賴

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-hive_2.11</artifactId>
    <version>2.1.1</version>
</dependency>
           

代碼實作

import org.apache.spark.sql.SparkSession


object HiveRead {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("HiveRead")
      //添加支援外置hive
      .enableHiveSupport()
      .getOrCreate()


    spark.sql("show databases")
    spark.sql("use guli")
    spark.sql("select count(*) from gulivideo_orc").show()


    spark.close()
  }
}
           

結果

import org.apache.spark.sql.SparkSession


object HiveWrite {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "root");
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("HiveRead")
      //添加支援外置hive
      .enableHiveSupport()
      .config("spark.sql.warehouse.dir","hdfs://hadoop102:9000/user/hive/warehouse")
      .getOrCreate()


    //先建立一個資料庫
    spark.sql("create database spark1602")
    spark.sql("use spark1602")
    spark.sql("create table user1(id int,name string)")
    spark.sql("insert into user1 values(10,'lisi')").show()


    spark.close()


  }
}
           

使用df.write.saveAsTable(“表名”)(常用)

import org.apache.spark.sql.SparkSession


object HiveWrite {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "root");
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("HiveRead")
      //添加支援外置hive
      .enableHiveSupport()
      .config("spark.sql.warehouse.dir","hdfs://hadoop102:9000/user/hive/warehouse")
      .getOrCreate()




    val df = spark.read.json("D:\\idea\\spark-sql\\input\\user.json")
    spark.sql("use spark1602")
    //直接把資料寫入到hive中,表可以存在也可以不存在
    df.write.saveAsTable("user2")
    //也可以進行追加
   //df.write.mode("append").saveAsTable("user2")
    spark.close()


  }
}
           

使用df.write.insertInto(“表名”)

import org.apache.spark.sql.SparkSession


object HiveWrite {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "root");
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("HiveRead")
      //添加支援外置hive
      .enableHiveSupport()
      .config("spark.sql.warehouse.dir","hdfs://hadoop102:9000/user/hive/warehouse")
      .getOrCreate()




    val df = spark.read.json("D:\\idea\\spark-sql\\input\\user.json")
    spark.sql("use spark1602")
    df.write.insertInto("user2")


    spark.close()
  }
}
           

saveAsTable

使用列名進行配置設定值

insertInto

按照位置進行1對1

import org.apache.spark.sql.SparkSession


object HiveWrite {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "root");
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("HiveRead")
      //添加支援外置hive
      .enableHiveSupport()
      .config("spark.sql.warehouse.dir","hdfs://hadoop102:9000/user/hive/warehouse")
      .getOrCreate()


    val df = spark.read.json("D:\\idea\\spark-sql\\input\\user.json")
    df.createOrReplaceTempView("a")


    spark.sql("use spark1602")
    val df1 = spark.sql("select * from a ")
    val df2 = spark.sql("select sum(age) sum_age from a group by name")
    println(df1.rdd.getNumPartitions)
    println(df2.rdd.getNumPartitions)
   df1.write.saveAsTable("a3")
   df2.write.saveAsTable("a4")


    spark.close()
  }
}
           

結果：聚合函數分區數預設200個

如果資料量小，沒必要200兩個分區，簡直浪費。

 df2.write.saveAsTable("a4")

修改為

 df2.coalesce(1).write.saveAsTable("a4")

版權聲明：

本文為大資料技術與架構整理，原作者獨家授權。未經原作者允許轉載追究侵權責任。

編輯｜譚林平丶

微信公衆号｜import_bigdata