object IPRuels {  // 将ip地址转换为整数  def ip2num(ip : String) : Long = {    val fragments = ip.split("\\.")    var ipNum = 0L    for (i <- 0 until fragments.length) {      // 与运算      ipNum = fragments(i).toLong | ipNum << 8L    }    ipNum  }  @volatile  private var instance : Broadcast[Array[(Long, Long, String)]] = null  def getInstance(sc: SparkContext): Broadcast[Array[(Long, Long, String)]] = {    if (instance == null) {      synchronized {        if (instance == null) {          val wordBlacklist = sc.textFile("/home/peerslee/spark_data/ip.txt").map(line =>            // map RDD 的Transformation 操作，用 f 处理一个Rdd 的所有元素，将结果输出到另一个Rdd          {            val fields = line.trim().split("\t")            val start_num = ip2num(fields(0).trim())            val end_num = ip2num(fields(1).trim())            val province = fields(2).trim()            (start_num, end_num, province)          }).collect()          instance = sc.broadcast(wordBlacklist)        }      }    }    instance  }}

def IPLocation(dStream: DStream[String], sc : SparkContext, mongoSQL: MongoSQL): Unit = {    val ipRulesBroadcast = IPRuels.getInstance(sc)    val result = dStream.map(line => { ipPat.findFirstIn(line.toString()).mkString("")})      .map(ip => {        var info : Any = None        if(!ip.isEmpty) {          val ipNum = ip2num(ip)          val index = binarySearch(ipRulesBroadcast.value, ipNum)          info = ipRulesBroadcast.value(index)        }        (info, 1L)})      .reduceByKey(_+_)    result.foreachRDD((rdd : RDD[(Any, Long)]) => {      // 将total 写在map,for里      var total = 1L      try {         total = rdd.reduce((x, y) => ("t", x._2 + y._2))._2      } catch {        case  ex : UnsupportedOperationException => {          total = 1        }      }      val rowRdd = rdd.map(x => {        val r = x._2.toFloat/total        (x._1, r)      }).map(line => Row(line._1.toString, line._2.toString))      val schemaStr = "loc,rate"      mongoSQL.put(schemaStr, rowRdd, "IP_rate_2")    })  }