利用spark实现ip地址查询

原始数据模板

用户浏览访问网络的日志模板

20090121000132095572000|125.213.100.123|show.51.com|/shoplist.php?phpfile=shoplist2.php&style=1&sex=137|Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; Mozilla/4.0(Compatible Mozilla/4.0(Compatible-EmbeddedWB 14.59 http://bsalsa.com/ EmbeddedWB- 14.59  from: http://bsalsa.com/ )|http://show.51.com/main.php|
20090121000132124542000|117.101.215.133|www.jiayuan.com|/19245971|Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; TencentTraveler 4.0)|http://photo.jiayuan.com/index.php?uidhash=d1c3b69e9b8355a5204474c749fb76ef|__tkist=0; myloc=50%7C5008; myage=2009; PROFILE=14469674%3A%E8%8B%A6%E6%B6%A9%E5%92%96%E5%95%A1%3Am%3Aphotos2.love21cn.com%2F45%2F1b%2F388111afac8195cc5d91ea286cdd%3A1%3A%3Ahttp%3A%2F%2Fimages.love21cn.com%2Fw4%2Fglobal%2Fi%2Fhykj_m.jpg; last_login_time=1232454068; SESSION_HASH=8176b100a84c9a095315f916d7fcbcf10021e3af; RAW_HASH=008a1bc48ff9ebafa3d5b4815edd04e9e7978050; COMMON_HASH=45388111afac8195cc5d91ea286cdd1b; pop_1232093956=1232468896968; pop_time=1232466715734; pop_1232245908=1232469069390; pop_1219903726=1232477601937; LOVESESSID=98b54794575bf547ea4b55e07efa2e9e; main_search:14469674=%7C%7C%7C00; registeruid=14469674; REG_URL_COOKIE=http%3A%2F%2Fphoto.jiayuan.com%2Fshowphoto.php%3Fuid_hash%3D0319bc5e33ba35755c30a9d88aaf46dc%26total%3D6%26p%3D5; click_count=0%2C3363619
20090121000132406516000|117.101.222.68|gg.xiaonei.com|/view.jsp?p=389|Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; CIBA)|http://home.xiaonei.com/Home.do?id=229670724|_r01_=1; __utma=204579609.31669176.1231940225.1232462740.1232467011.145; __utmz=204579609.1231940225.1.1.utmccn=(direct)
20090121000132581311000|115.120.36.118|tj.tt98.com|/tj.htm|Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; TheWorld)|http://www.tt98.com/|
20090121000132864647000|123.197.64.247|cul.sohu.com|/20071227/n254338813_22.shtml|Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; TheWorld)|http://cul.sohu.com/20071227/n254338813_22.shtml|ArticleTab=visit:1; IPLOC=unknown; SUV=0901080709152121; vjuids=832dd37a1.11ebbc5d590.0.b20f858f14e918; club_chat_ircnick=JaabvxC4aaacQ; spanel=%7B%22u%22%3A%22%22%7D; vjlast=1232467312,1232467312,30
20090121000133296729000|222.55.57.176|down.chinaz.com|/|Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; iCafeMedia; TencentTraveler 4.0)||cnzz_a33219=0; vw33219=%3A18167791%3A; sin33219=http%3A//www.itxls.com/wz/wyfx/it.html; rtime=0; ltime=1232464387281; cnzz_eid=6264952-1232464379-http%3A//www.itxls.com/wz/wyfx/it.html
20090121000133331104000|123.197.66.93|www.pkwutai.cn|/down/downLoad-id-45383.html|Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; QQDownload 1.7)|http://www.baidu.com/s?tn=b1ank_pg&ie=gb2312&bs=%C3%C0%C6%BC%B7%FE%D7%B0%B9%DC%C0%ED%C8%ED%BC%FE&sr=&z=&cl=3&f=8&wd=%C6%C6%BD%E2%C3%C0%C6%BC%B7%FE%D7%B0%B9%DC%C0%ED%C8%ED%BC%FE&ct=0|
20090121000133446262000|115.120.12.157|v.ifeng.com|/live/|Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1) ; .NET CLR 2.0.50727; CIBA)|http://www.ifeng.com/|userid=1232466610953_4339; location=186; sclocationid=10002; vjuids=22644b162.11ef4bc1624.0.63ad06717b426; vjlast=1232466614,1232467297,13
20090121000133456256000|115.120.7.240|cqbbs.soufun.com|/3110502342~-1~2118/23004348_23004348.htm|Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1) ; .NET CLR 2.0.50727; CIBA)||new_historysignlist=%u534E%u6DA6%u4E8C%u5341%u56DB%u57CE%7Chttp%3A//cqbbs.soufun.com/board/3110502342/%7C%7C%u9A8F%u9038%u7B2C%u4E00%u6C5F%u5CB8%7Chttp%3A//cqbbs.soufun.com/board/3110169184/%7C%7C%u793E%u533A%u4E4B%u661F%7Chttp%3A//cqbbs.soufun.com/board/sqzx/%7C%7C; SoufunSessionID=2y5xyr45kslc0zbdooqnoo55; viewUser=1; vjuids=-870e9088.11ee89aba57.0.be9c3d988def8; vjlast=1232263101,1232380806,11; new_viewtype=1; articlecolor=#000000; usersms_pop_type=1; articlecount=186; __utma=101868291.755195653.1232450942.1232450942.1232450942.1; __utmz=101868291.1232450942.1.1.utmccn=(referral)
20090121000133586141000|117.101.219.241|12.zgwow.com|/launcher/index.htm|Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1)||

IP地址的映射模板

1.0.1.0|1.0.3.255|16777472|16778239|亚洲|中国|福建|福州||电信|350100|China|CN|119.306239|26.075302
1.0.8.0|1.0.15.255|16779264|16781311|亚洲|中国|广东|广州||电信|440100|China|CN|113.280637|23.125178
1.0.32.0|1.0.63.255|16785408|16793599|亚洲|中国|广东|广州||电信|440100|China|CN|113.280637|23.125178
1.1.0.0|1.1.0.255|16842752|16843007|亚洲|中国|福建|福州||电信|350100|China|CN|119.306239|26.075302
1.1.2.0|1.1.7.255|16843264|16844799|亚洲|中国|福建|福州||电信|350100|China|CN|119.306239|26.075302
1.1.8.0|1.1.63.255|16844800|16859135|亚洲|中国|广东|广州||电信|440100|China|CN|113.280637|23.125178
1.2.0.0|1.2.1.255|16908288|16908799|亚洲|中国|福建|福州||电信|350100|China|CN|119.306239|26.075302
1.2.2.0|1.2.2.255|16908800|16909055|亚洲|中国|北京|北京|海淀|北龙中网|110108|China|CN|116.29812|39.95931
1.2.4.0|1.2.4.255|16909312|16909567|亚洲|中国|北京|北京||中国互联网信息中心|110100|China|CN|116.405285|39.904989
1.2.5.0|1.2.7.255|16909568|16910335|亚洲|中国|福建|福州||电信|350100|China|CN|119.306239|26.075302

需求

  根据用户访问网络的IP地址查找到用户所在地

实现过程

1. 加载城市ip段信息，获取ip起始数字和结束数字，经度，维度
2. 加载日志数据，获取ip信息，然后转换为数字，和ip段比较
3. 比较的时候采用二分法查找，找到对应的经度和维度
4. 对精度和维度做单词计数

代码

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

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

//todo：利用spark实现IP地址查询
object IpLocation {

  //将ip地址转化为Long
  def ip2Long(ip: String): Long = {
    //固定的算法
    val ips : Array[String] = ip.split("\\.")
    var ipNum:Long = 0L
    for (i <- ips){
      ipNum = i.toLong | ipNum <<8L
    }
    ipNum
  }

  //利用二分法查询，查询到long类型的数字在数组的下标
  def binarySearch(ipNum: Long, broadcastValue: Array[(String, String, String, String)]): Int = {
      //开始下标
      var start = 0
      //结束下标
      var end = broadcastValue.length-1

      while(start <= end){
          var middle = (start+end)/2
          if(ipNum>=broadcastValue(middle)._1.toLong && ipNum<=broadcastValue(middle)._2.toLong){
              return middle
          }
          if(ipNum < broadcastValue(middle)._1.toLong){
              end=middle
          }
          if(ipNum > broadcastValue(middle)._2.toLong){
            start=middle
          }
      }
    -1
  }

  //将数据写入mysql表中
  def data2mysql(iter:Iterator[((String,String),Int)])={
      //定义数据库链接
      var conn:Connection = null
      //定义PreparedStatement
      var ps : PreparedStatement=null
      //编写sql语句
      val sql="insert into iplocation(longitude,latitude,total_count) values(?,?,?)"
      //获取数据库连接
      conn=DriverManager.getConnection("jdbc:mysql://192.168.58.200:3306/spark","root","123456")
      ps=conn.prepareStatement(sql)

      //遍历迭代器
      iter.foreach(line=>{
        //对sql语句中的占位符赋值
        ps.setString(1,line._1._1)
        ps.setString(2,line._1._2)
        ps.setInt(3,line._2)
        //执行sql语句
        ps.execute()
      })
  }

  def main(args: Array[String]): Unit = {
    //1.创建sparkConf对象
    val sparkConf = new SparkConf().setAppName("IpLocation").setMaster("local[2]")
    //2.创建sparkContext对象
    val sc = new SparkContext(sparkConf)
    //3.读取城市ip信息，获取ip开始数字、ip结束数字、经度、维度
    val city_ip_rdd : RDD[(String,String,String,String)]= sc.textFile("f:\\text\\ip.txt").map(_.split("\\|")).map(x=>(x(2),x(3),x(13),x(14)))
    //将城市ip信息广播到每一个worker节点
    val cityIpBroadcast = sc.broadcast(city_ip_rdd.collect())
    //4.获取日志数据,获取所有ip地址
    val destData = sc.textFile("f:\\text\\20090121000132.394251.http.format").map(_.split("\\|")(1))
    //5.遍历所有IP地址，去广播变量中进行匹配，获取对应的经度和纬度
    val result: RDD[((String, String), Int)] = destData.mapPartitions(iter => {
      //获取广播变量的值
      val broadcastValue: Array[(String, String, String, String)] = cityIpBroadcast.value
      //遍历迭代器，获取每一个ip地址
      iter.map(ip => {
        //需要将ip地址转化为long类型
        val ipNum: Long = ip2Long(ip)
        //拿到long类型数字去广播变量中进行匹配，利用二分查询
        val index: Int = binarySearch(ipNum, broadcastValue)
        //返回结果数据((经度，维度)，1)
        ((broadcastValue(index)._3, broadcastValue(index)._4), 1)
      })
    })
    //6.把相同经度和维度出现的次数相加
    val finalResult: RDD[((String, String), Int)] = result.reduceByKey(_+_)
    //7.打印输出结果
    finalResult.collect().foreach(x=>println(x))

    //保存结果数据到mysql中
    finalResult.foreachPartition(data2mysql)

    //8.关闭sc
    sc.stop()
  }
}

知识点总结：

• 数据的广播：由于我们在处理数据的时候需要将数据分配到多个task上来执行，如果每一个task都存放一个ip的信息进行匹配，会占用我们大量的内存，因此spark提供了广播数据的功能，将共同使用的数据在每一个worker中存放一份，当task使用的时候从worker中调用即可，这样就减少了内存的占用。
• map和mapPartitions的区别：map作用在每一个元素，而mapPartitions作用在每一个分区上