Spark实战案例二, 通过Spark实现IP地址查询,广播大变量broadcast 16

- 日理万妓 2021-11-29 07:14 335阅读 0赞

# 1. 需求分析 #

在互联网中,我们经常会见到城市热点图这样的报表数据,例如在百度统计中, 会统计今年的热门旅游城市, 热门包括学校等.会将这样的信息显示在热点图中.  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70]  
因此,我们需要通过日志信息(运营商或者网站自己生产)和城市ip段信息来判断用户的ip段,统计热点经纬度.

# 2. 技术调研 #

因为我们的需求是完成一张报表信息,所以对程序的实时性没有要求,所以可以选择内存计算spark来实现上述功能.

# 3. 结构设计 #

搭建spark集群, 详解博主播客,链接如下[Spark集群安装,Web界面安装][Spark_Web]

# 4. 开发流程 #

## 4.1 数据准备 ##

## 4.2 ip日志信息 ##

在ip日志信息中,我们只关系ip这一维度就可以了,其他的不做介绍  
![在这里插入图片描述][20190721160954120.png]

## 4.3 城市ip段信息 ##

![在这里插入图片描述][20190721161123427.png]

# 5. 代码开发 #

## 5.1 思路 ##

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

## 5.2 代码 ##

package cn.IPlocation
    
    import java.sql.{ Connection,DriverManager, PrepareStatement}
    import org.apache.spark.broadcast.Broadcast
    import org.apache.spark.rdd.RDD
    import org.apache.spark.{ SparkConf, SparkContext}
    
    object IPlocation_Test{ 
    	def main(args: Array[String]): Unit = { 
    		//创建sparkConf, 设置参数
    		val sparkConf: SparkConf = new SparkConf().setAppName("IPlocation_Test").setMaster("local[2]")
    		//创建SparkContext
    		val sc = new SparkContext(sparkConf)
    		//读取基站数据
    		val data: RDD[String] = sc.textFile("d:\\data\\ip.txt")
    		//对基站数据进行切分, 获取需要的字段(ipStart, ipEnd, 城市位置,经度,纬度)
    		val jizhangRDD: RDD[(String,String,String,String,String)] = data.map(_.split("\\|")).map(x => (x(2), x(3), x(4) + "-" + x(5) + "-" + x(6) + "-" + x(7) + "-" + x(8), x(13), x(14)))
    		//获取RDD的数据
    		val jizhanData: Array[(String, String, String, String, String)] = jizhanRDD.collect()
    		//广播变量, 一个只读的输进去,所有的task都能读到的地方
    		val jizhanBroadcast: Broadcast[Array[(String, String, String, String, String)]] = sc.broadcast(jizhanData)
    		//读取目标数据
    		val destData:RDD[String] = sc.textFile("d:\\data\\20090121000132.394251.http.format")
    		//获取数据中的ip地址字段
    		val ipData: RDD[String] = destData.map(_.split("\\|")).map(x=>x(1))
    		//把ip地址转化为long类型, 然后通过二分法去基站数据中查找, 找到的维度做wordCount
    		val result=ipData.mapPartitions(iter=>{ 
          //获取广播变量中的值
          val valueArr: Array[(String, String, String, String, String)] = jizhanBroadcast.value
    
          //todo:操作分区中的itertator
          iter.map(ip=>{ 
            //将ip转化为数字long
            val ipNum:Long=ipToLong(ip)
    
            //拿这个数字long去基站数据中通过二分法查找，返回ip在valueArr中的下标
            val index:Int=binarySearch(ipNum,valueArr)
    
            //根据下标获取对一个的经纬度
            val tuple = valueArr(index)
            //返回结果 ((经度，维度)，1)
            ((tuple._4,tuple._5),1)
    
          })
    
        })
    
        //todo:分组聚合
        val resultFinal: RDD[((String, String), Int)] = result.reduceByKey(_+_)
    
        //todo:打印输出
        resultFinal.foreach(println)
    
        //todo:将结果保存到mysql表中
    
    resultFinal.map(x=>(x._1._1,x._1._2,x._2)).foreachPartition(data2Mysql)
    sc.stop()
    
      }
    
      //todo：ip转为long类型
      def ipToLong(ip: String): Long = { 
        //todo：切分ip地址。
        val ipArray: Array[String] = ip.split("\\.")
        var ipNum=0L
    
        for(i <- ipArray){ 
          ipNum=i.toLong | ipNum << 8L
        }
        ipNum
      }
    
      //todo:通过二分查找法,获取ip在广播变量中的下标
      def binarySearch(ipNum: Long, valueArr: Array[(String, String, String, String, String)]): Int ={ 
        //todo:口诀：上下循环寻上下，左移右移寻中间
        //开始下标
        var start=0
        //结束下标
        var end=valueArr.length-1
    
        while(start<=end){ 
          val middle=(start+end)/2
          if(ipNum>=valueArr(middle)._1.toLong && ipNum<=valueArr(middle)._2.toLong){ 
            return middle
          }
    
          if(ipNum > valueArr(middle)._2.toLong){ 
            start=middle
          }
    
          if(ipNum<valueArr(middle)._1.toLong){ 
            end=middle
          }
        }
    
        -1
      }
    
      //todo:数据保存到mysql表中
      def data2Mysql(iterator:Iterator[(String,String, Int)]):Unit = { 
        //todo:创建数据库连接Connection
        var conn:Connection=null
        //todo:创建PreparedStatement对象
        var ps:PreparedStatement=null
        //todo:采用拼占位符问号的方式写sql语句。
        var sql="insert into iplocaltion(longitude,latitude,total_count) values(?,?,?)"
        //todo:获取数据连接
        conn=DriverManager.getConnection("jdbc:mysql://itcast01:3306/spark","root","root123")
    
        //todo: 选中想被try/catch包围的语句 ctrl+alt+t 快捷键选中try/catch/finally
        try { 
            iterator.foreach(line=> { 
    
            //todo:预编译sql语句
            ps = conn.prepareStatement(sql)
    
            //todo:对占位符设置值，占位符顺序从1开始，第一个参数是占位符的位置，第二个参数是占位符的值。
            ps.setString(1, line._1)
            ps.setString(2, line._2)
            ps.setLong(3, line._3)
            //todo:执行
            ps.execute()
            })
          } catch { 
            case e:Exception =>println(e)
          } finally { 
            if(ps!=null){ 
              ps.close()
            }
            if (conn!=null){ 
              conn.close()
            }
          }
      }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70]: /images/20211129/1f7e0c75eec74e4fa0bec6b91d3ca399.png
[Spark_Web]: https://blog.csdn.net/weixin_42641909/article/details/90671137
[20190721160954120.png]: /images/20211129/4bfeb64f55a04d419c61d6acb64fc352.png
[20190721161123427.png]: /images/20211129/490333c962b14059a44385ea1ff135c5.png