?

以前我用spark連接hive都是把hive的配置文件放在spark的conf目錄，后來有個項目在運行時才能確定要連接哪個hive源，我就找了個能在運行時加載配置文件的方法，代碼如下：

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.spark.sql.SparkSession
import scala.collection.JavaConverters._
?
object ReadHive {
?
? def main(args: Array[String]): Unit = {
? ? val sparkBuilder = SparkSession
? ? ? .builder
? ? ? .master("local")
? ? ? .appName("Spk Pi")
? ? val conf = new Configuration()
? ? // 這里的文件地址可以換成從數據庫里查詢
? ? val core = new Path("C:\\Users\\shadow\\Desktop\\core-site.xml")
? ? val hdfs = new Path("C:\\Users\\shadow\\Desktop\\hdfs-site.xml")
? ? val hive = new Path("C:\\Users\\shadow\\Desktop\\hive-site.xml")
? ? conf.addResource(core)
? ? conf.addResource(hdfs)
? ? conf.addResource(hive)
? ? for (c <- conf.iterator().asScala){
? ? ? sparkBuilder.config(c.getKey, c.getValue)
? ? }
? ? val spark = sparkBuilder.enableHiveSupport().getOrCreate()
? ? spark.sql("select * from default.wt_test1").show()
? }
?
}
?
?

我們以wordcount為例，分析如何配置。我們的輸入數據源來自cluster1的HDFS，需要將分析結果輸出到cluster2的HDFS。

val conf = new SparkConf().setAppName("Spark Word Count") val sc = new SparkContext()// 在輸入數據之前先將hadoop config配置為cluster1集群 sc.hadoopConfiguration.addResource("cluster1/core-site.xml") sc.hadoopConfiguration.addResource("cluster1/hdfs-site.xml")// load data val input = sc.textFile(args(0)).flatMap(_.split(" ")).map(x => (x, 1)).reduceByKey(_ + _)// 再將hadoop config設為cluster2集群 sc.hadoopConfiguration.addResource("cluster2/core-site.xml") sc.hadoopConfiguration.addResource("cluster2/hdfs-site.xml") input.saveAsTextFile(args(1))

core-site.xml和hdfs-site.xml放在項目的resources目錄下

通過上述例子我們可以看到，我們如果需要在spark任務中想操作不同的hdfs集群，我們需要在操作之前先將hadoop的config設置為我們需要操作的目標HDFS集群即可。

向spark提交任務：

bin/spark-submit --master yarn-client --class SparkWordcount run.jar /input /output

NOTE: 這里我們即可以寫成全路徑形式，即：hdfs://cluster1/input hdfs://cluster2/output，也可以寫成上面相對路徑的形式。

上面我們通過hadoopConfiguration的addResource方法來添加相關配置，其實Spark在操作hdfs的時候，只需hadoop的ha相關配置就可以了，所以我們也可以通過代碼來直接配置hadoop的相關配置。

val conf = new SparkConf().setAppName("Spark Word Count")val sc = new SparkContext()sc.hadoopConfiguration.set("fs.defaultFS", "hdfs://cluster1");sc.hadoopConfiguration.set("dfs.nameservices", "cluster1");sc.hadoopConfiguration.set("dfs.ha.namenodes.cluster1", "nn1,nn2");sc.hadoopConfiguration.set("dfs.namenode.rpc-address.cluster1.nn1", "namenode001:8020");sc.hadoopConfiguration.set("dfs.namenode.rpc-address.cluster1.nn2", "namenode002:8020");sc.hadoopConfiguration.set("dfs.client.failover.proxy.provider.cluster1", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");val wc = sc.textFile(args(0)).flatMap(_.split(" ")).map(x => (x, 1)).reduceByKey(_ + _)sc.hadoopConfiguration.set("fs.defaultFS", "hdfs://cluster2");sc.hadoopConfiguration.set("dfs.nameservices", "cluster2");sc.hadoopConfiguration.set("dfs.ha.namenodes.cluster2", "nn3,nn4");sc.hadoopConfiguration.set("dfs.namenode.rpc-address.cluster2.nn3", "namenode003:8020");sc.hadoopConfiguration.set("dfs.namenode.rpc-address.cluster2.nn4", "namenode004:8020");sc.hadoopConfiguration.set("dfs.client.failover.proxy.provider.cluster2", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");wc.saveAsTextFile(args(1))

這樣我們就兩種不同的方式來配置hadoop的config，我們可以根據自己的需求來選擇需要用哪種方式來配置

?

總結

以上是生活随笔為你收集整理的spark运行时加载hive，hdfs配置文件的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。