之前案例三中的ip地址規則是在Driver端的機器磁碟中儲存著的，但是現在如果實在hdfs中儲存著的又該如何實現呢

首先要分析清楚才能實現，儲存在hdfs中並不像想象中的那麼容易，首先程式碼實在Driver端寫的，在Driver端寫從hdfs中取出ip地址規則的程式碼會觸發action，然後生成Task，分發到Executor端執行，因為日誌檔案比較大，所以儲存在hdfs中的時候是會分割槽儲存的，這樣每個Executor只能取到一部分的ip地址規則了，所以這時候應該將所有Executor中的部分ip地址規則收集到Driver端，然後再由Driver端進行廣播，Executor使用廣播變數的引用來取到完整的ip地址規則，其他的實現則跟案例三中的一樣

具體實現程式碼：

package cn.ysjh0014

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object IpLocation2 {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("IpLoaction1").setMaster("local[4]")

    val sc = new SparkContext(conf)

    //取到HDFS中的ip規則
    val rulesLines: RDD[String] = sc.textFile(args(0))
    //整理ip規則資料
    val ipRulesRDD: RDD[(Long, Long, String)] = rulesLines.map(line => {
      val fields = line.split("[|]")
      val startNum = fields(2).toLong
      val endNum = fields(3).toLong
      val province = fields(6)
      (startNum, endNum, province)
    })

    //將分散在多個Executor中的部分IP規則收集到Driver端
    val rulesInDriver: Array[(Long, Long, String)] = ipRulesRDD.collect()

    //將Driver端的資料廣播到Executor
    //廣播變數的引用（還在Driver端）
    val broadcastRef: Broadcast[Array[(Long, Long, String)]] = sc.broadcast(rulesInDriver)

    //建立RDD，讀取訪問日誌
    val accessLines: RDD[String] = sc.textFile(args(1))

    //整理資料
    val proviceAndOne: RDD[(String, Int)] = accessLines.map(log => {
      //將log日誌的每一行進行切分
      val fields = log.split("[|]")
      val ip = fields(1)
      //將ip轉換成十進位制
      val ipNum = TestIp.ip2Long(ip)
      //進行二分法查詢，通過Driver端的引用或取到Executor中的廣播變數
      //（該函式中的程式碼是在Executor中別調用執行的，通過廣播變數的引用，就可以拿到當前Executor中的廣播的規則了）
      //Driver端廣播變數的引用是怎樣跑到Executor中的呢？
      //Task是在Driver端生成的，廣播變數的引用是伴隨著Task被髮送到Executor中的
      val rulesInExecutor: Array[(Long, Long, String)] = broadcastRef.value
      //查詢
      var province = "未知"
      val index = TestIp.binarySearch(rulesInExecutor, ipNum)
      if (index != -1) {
        province = rulesInExecutor(index)._3
      }
      (province, 1)
    })

    //聚合
    //val sum = (x: Int, y: Int) => x + y
    val reduced: RDD[(String, Int)] = proviceAndOne.reduceByKey(_ + _)

    def data2MySQL(it: Iterator[(String, Int)]): Unit = {
      //一個迭代器代表一個分割槽，分割槽中有多條資料
      //先獲得一個JDBC連線
      val conn: Connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8", "root", "root")
      //將資料通過Connection寫入到資料庫
      val pstm: PreparedStatement = conn.prepareStatement("INSERT INTO access_log VALUES (?, ?)")
      //將分割槽中的資料一條一條寫入到MySQL中
      it.foreach(tp => {
        pstm.setString(1, tp._1)
        pstm.setInt(2, tp._2)
        pstm.executeUpdate()
      })
      //將分割槽中的資料全部寫完之後，在關閉連線
      if (pstm != null) {
        pstm.close()
      }
      if (conn != null) {
        conn.close()
      }
    }


    reduced.foreachPartition(it => data2MySQL(it))


    sc.stop()

  }
}
 

執行結果跟案例三中的一樣