點擊上方藍(lán)色字體，選擇“設(shè)為星標(biāo)”

回復(fù)”資源“獲取更多資源

點擊右側(cè)關(guān)注，大數(shù)據(jù)開發(fā)領(lǐng)域最強公眾號！

點擊右側(cè)關(guān)注，暴走大數(shù)據(jù)！

一.hive和spark sql的集成方式(面試可能會問到)

hive on spark(版本兼容)
官網(wǎng)https://cwiki.apache.org/confluence/display/Hive/Hive+on+Spark%3A+Getting+Started

spark on hive(版本兼容)
官網(wǎng)
http://spark.apache.org/docs/2.1.1/sql-programming-guide.html#hive-tables

二.spark_shell和spark_sql操作

spark_shell

如果你在集群上使用了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

spark_sql

完全跟sql一樣

使用hiveserver2 + beeline

spark-sql 得到的結(jié)果不夠友好, 所以可以使用hiveserver2 + beeline
1.啟動thriftserver(后臺)

sbin/start-thriftserver.sh

2.啟動beeline

bin/beeline# 然后輸入!connect jdbc:hive2://hadoop102:10000#?然后按照提示輸入用戶名和密碼

三.腳本使用spark-sql

四.idea中讀寫Hive數(shù)據(jù)

1.從hive中讀數(shù)據(jù)

添加依賴

    org.apache.spark    spark-hive_2.11    2.1.1

代碼實現(xiàn)

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()??}}

結(jié)果

2.從hive中寫數(shù)據(jù)

使用hive的insert語句去寫

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()
    //先創(chuàng)建一個數(shù)據(jù)庫    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")    //直接把數(shù)據(jù)寫入到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()??}}

3.saveAsTable和insertInto的原理

saveAsTable
使用列名進行分配值

insertInto
按照位置進行1對1

五.聚合后的分區(qū)數(shù)

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()??}}

結(jié)果：聚合函數(shù)分區(qū)數(shù)默認(rèn)200個

如果數(shù)據(jù)量小，沒必要200兩個分區(qū)，簡直浪費。

?df2.write.saveAsTable("a4")

修改為

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

版權(quán)聲明：

本文為大數(shù)據(jù)技術(shù)與架構(gòu)整理，原作者獨家授權(quán)。未經(jīng)原作者允許轉(zhuǎn)載追究侵權(quán)責(zé)任。

編輯｜譚林平丶

微信公眾號｜import_bigdata

歡迎點贊+收藏+轉(zhuǎn)發(fā)朋友圈素質(zhì)三連

文章不錯？點個【在看】吧！??

Spark SQL快速入門系列之Hive

目錄