spark与Hive的整合入门

SparkSQL和Hive的整合，是一种比较常见的关联处理方式，SparkSQL加载Hive中的数据进行业务处理，同时将计算结果落地回Hive中。

example

idea新建项目,并把hive-site.xml,core-site.xml,hdfs-site.xml文件下载到Resources文件夹下.

读取数据

object HiveDemo extends App{private val session: SparkSession = SparkSession.builder().enableHiveSupport().master("local").appName("hive").getOrCreate()//Returns the specified table/view as a DataFrame.//将hive的table转为df对象private val frame: DataFrame = session.table("sz.emp1")//查询每个部门的平均工资,最高工资,最低工资,总人数//创建临时视图frame.createTempView("a")val sql="""|select deptno,max(sal),min(sal),avg(nvl(sal,0))|from|a|group by deptno|""".stripMargin//使用Spark执行SQL查询，以数df形式返回结果.session.sql(sql).show()
}

结果如下

+------+--------+--------+--------------------+
|deptno|max(sal)|min(sal)|avg(nvl(a.`sal`, 0))|
+------+--------+--------+--------------------+
|    20|    3000|     800|              2175.0|
|    10|    5000|    1300|  2916.6666666666665|
|    30|    2850|     950|  1566.6666666666667|
+------+--------+--------+--------------------+

也可以如下查询
agg方法:通过指定列名和聚合方法计算聚合。结果DataFrame还将包含分组列。
可用的聚合方法有avg、max、min、sum、count。

frame.groupBy("deptno").agg("sal"->"max","sal"->"min").show()

结果如下

+------+--------+--------+
|deptno|max(sal)|min(sal)|
+------+--------+--------+
|    20|    3000|     800|
|    10|    5000|    1300|
|    30|    2850|     950|
+------+--------+--------+

写入数据

saveAsTable方法
将DataFrame的内容另存为指定的表。
在表已经存在的情况下，此函数的行为取决于由mode函数指定的保存模式（默认为引发异常）。当mode为Overwrite时，DataFrame的架构不需要与现有表的架构相同。
当mode为Append时，如果有一个现有表，我们将使用现有表的格式和选项。表中的数据顺序不必与该列的数据顺序相同。与InsertInto不同，saveAsTable将使用列名来查找正确的列位置。

object HiveDemo2 extends App {System.setProperty("HADOOP_USER_NAME", "root")private val session: SparkSession = SparkSession.builder().enableHiveSupport()//往动态分区表写数据需要这个.config("hive.exec.dynamic.partition.mode", "nonstrict").master("local").appName("hive").getOrCreate()//连接hive表返回df对象private val frame: DataFrame = session.table("sz.emp1")//存入数据,如果没有,会在hive中新建表格frame.write.saveAsTable("sz.temp1")session.stop()
}

例子

scala> Seq((1, 2)).toDF("i", "j").write.mode("overwrite").saveAsTable("t3")
//saveAsTable会基于列名来解析数据
scala> Seq((3, 4)).toDF("j", "i").write.mode("append").saveAsTable("t3")scala> sql("select * from t3").show
+---+---+
|  i|  j|
+---+---+
|  1|  2|
|  4|  3|
+---+---+

insertInto方法
将DataFrame的内容插入到指定的表中。它要求DataFrame的schema与表的schema相同。与saveAsTable不同，insertInto忽略列名，只使用基于位置的解析。

object HiveDemo3 extends App {System.setProperty("HADOOP_USER_NAME", "root")private val session: SparkSession = SparkSession.builder().enableHiveSupport().master("local").appName("hive").getOrCreate()//读取emp1表的数据转为DF对象private val frame: DataFrame = session.table("sz.emp1")//将上述DF对象的数据保存到temp1表中. 模式为追加模式.要求DataFrame的schema与表的schema相同.frame.write.mode(SaveMode.Append).insertInto("sz.temp1")session.stop()
}

再比如

//向表t2中写入数据,如不存在,则创建
scala>
Seq((1,2)).toDF("i","j").write.mode("overwrite").saveAsTable("t2")
//insertInto只基于位置的解析,与名字无关
scala> Seq((3,4)).toDF("j","i").write.insertInto("t2")
//insertInto只基于位置的解析,与名字无关
scala> Seq((5,6)).toDF("a","b").write.insertInto("t2")
//查看结果
scala> sql("select * from t2").show
+---+---+
|  i|  j|
+---+---+
|  1|  2|
|  3|  4|
|  5|  6|
+---+---+

错误解决

如果报权
限错误,在hive所在机器执行如下进行授权

 hdfs dfs -chmod -R 777 /

总结

读取数据,可以创建临时视图.利用sql方法查询
写入数据,可以用saveAsTable方法或者insertInto方法. 其中saveAsTable会基于列名解析,而insertInto方法仅仅基于位置进行解析