spark連接hive

dongkelun.com

1、在伺服器（虛擬機）spark-shell連接hive

1.1 將hive-site.xml拷貝到spark/conf里

cp /opt/apache-hive-2.3.2-bin/conf/hive-site.xml /opt/spark-2.2.1-bin-hadoop2.7/conf/

1.2 將mysql驅動拷貝到spark/jar里

cp /opt/apache-hive-2.3.2-bin/bin/mysql-connector-java-5.1.46-bin.jar /opt/spark-2.2.1-bin-hadoop2.7/jars/

1.3 啟動spark-shell,輸入代碼測試

spark-shellimport org.apache.spark.sql.hive.HiveContextval hiveContext = new HiveContext(sc)hiveContext.sql("select * from test").show()

1.4 異常及解決

在執行hiveContext.sql("select * from test").show() 報了一個異常：

The root scratch dir: /tmp/hive on HDFS should be writable. Current permissions are: rwxr-xr-x;

解決辦法：

1.4.1 更改HDFS目錄/tmp/hive的許可權：

hadoop fs -chmod 777 /tmp/hive

1.4.2 同時刪HDFS與本地的目錄/tmp/hive：

hadoop fs -rm -r /tmp/hive rm -rf /tmp/hive

這次錯誤採用的是第二種解決辦法，有的情況下用第一種方法，比如一次在啟動hive時候報這種錯誤~。 錯誤截圖：

參考：http://www.cnblogs.com/czm1032851561/p/5751722.html

2、win10+eclipse上連接hive

2.1 將hive-site.xml拷貝到項目中的resources文件夾下

2.2 在sbt里添加對應版本的mysql依賴

"mysql" % "mysql-connector-java" % "5.1.46"

2.3 代碼

2.3.1 舊版api(1.6以上)

package com.dkl.leanring.spark.sqlimport org.apache.spark.SparkConfimport org.apache.spark.sql.SQLContextimport org.apache.spark.sql.hive.HiveContextimport org.apache.spark.SparkContext/** * 舊版本spark-hive測試 */object OldSparkHiveDemo { def main(args: Array[String]): Unit = { val conf = new SparkConf().setAppName("OldSparkHiveDemo").setMaster("local") val sc = new SparkContext(conf) val sqlContext = new SQLContext(sc) import sqlContext.implicits._ val hiveCtx = new HiveContext(sc) hiveCtx.sql("select * from test").show() val data = Array((3, "name3"), (4, "name4"), (5, "name5")) val df = sc.parallelize(data).toDF("id", "name") df.createOrReplaceTempView("user") hiveCtx.sql("insert into test select id,name from user") hiveCtx.sql("select * from test").show() }}

（註：其中df.createOrReplaceTempView("user")改為df.registerTempTable("user")，因為createOrReplaceTempView方法是2.0.0才有的，registerTempTable是舊版的方法，1.6.0就有了，嫌麻煩就不改代碼重新貼圖了）

2.3.2 新版api

package com.dkl.leanring.spark.sqlimport org.apache.spark.sql.SparkSession/** * 新版本spark-hive測試 */object NewSparkHiveDemo { def main(args: Array[String]): Unit = { val spark = SparkSession .builder() .appName("Spark Hive Example") .master("local") .config("spark.sql.warehouse.dir", "/user/hive/warehouse/") .enableHiveSupport() .getOrCreate() import spark.implicits._ import spark.sql sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)") val data = Array((1, "val1"), (2, "val2"), (3, "val3")) var df = spark.createDataFrame(data).toDF("key", "value") df.createOrReplaceTempView("temp_src") sql("insert into src select key,value from temp_src") sql("SELECT * FROM src").show() }}

2.4 異常及解決方法

在執行insert語句時會出現如下異常信息：

org.apache.hadoop.security.AccessControlException: Permission denied: user=dongkelun, access=EXECUTE, inode="/user/hive/warehouse":root...

原因是：啟動 Spark 應用程序的win用戶對spark.sql.warehouse.dir沒有寫許可權 解決辦法：

hadoop fs -chmod 777 /user/hive/warehouse/

附異常信息截圖：

3、更新：寫hive方法

直接用下面這句代碼即可將df里的全部數據存到hive表裡

df.write.mode(SaveMode.Append).saveAsTable(tableName)

推薦閱讀：