spark练习---ip匹配以及广播的特性

package cn.wj.spark.day04

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by WJ on 2017/1/4.
  */
object IPLocation {

  //将IP转换成为Long类型
  def ip2Long(ip: String): Long = {
    val fragments = ip.split("[.]")
    var ipNum = 0L
    for (i <- 0 until fragments.length){
      ipNum =  fragments(i).toLong | ipNum << 8L
    }
    ipNum
  }

  //使用二分法，对IP进行查找，让ip与start_num以及end_num做对比
  def binarySearch(lines: Array[(String, String, String)], ip: Long) : Int = {
    var low = 0
    var high = lines.length - 1
    while (low <= high) {
      val middle = (low + high) / 2
      if ((ip >= lines(middle)._1.toLong) && (ip <= lines(middle)._2.toLong))
        return middle
      if (ip < lines(middle)._1.toLong)
        high = middle - 1
      else {
        low = middle + 1
      }
    }
    -1
  }

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("IPLocation")
    val sc = new SparkContext(conf)

    val ipRulesRdd = sc.textFile("e://Test/ip.txt").map(lines =>{
      val fields = lines.split("\\|")
      val start_num = fields(2)
      val end_num = fields(3)
      val province = fields(6)
      (start_num,end_num,province)
    })
    //全部的IP映射规则
    val ipRulesArrary = ipRulesRdd.collect()

    //广播规则，这个是由Driver向worker中广播规则
    val ipRulesBroadcast = sc.broadcast(ipRulesArrary)

    //加载要处理的数据
    val ipsRDD = sc.textFile("e://Test/access_log").map(line =>{
      val fields = line.split("\\|")
      fields(1)
    })

    val result = ipsRDD.map(ip =>{
      val ipNum = ip2Long(ip)
      val index = binarySearch(ipRulesBroadcast.value,ipNum)
      val info = ipRulesBroadcast.value(index)
      info
    })

    println(result.collect().toBuffer)
    sc.stop()
  }
}

 //将IP转换成为Long类型
  def ip2Long(ip: String): Long = {
    val fragments = ip.split("[.]")
    var ipNum = 0L
    for (i <- 0 until fragments.length){
      ipNum =  fragments(i).toLong | ipNum << 8L
    }
    ipNum
  }