以前我用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，我們可以根據自己的需求來選擇需要用哪種方式來配置