【Spark】如何用Spark查询IP地址？

【Spark】如何用Spark查询IP地址？

文章目录

• • 需求
  • 思路
  • ip地址转换为Long类型的两种方法
  • • ip地址转换数字地址的原理
    • 第一种方法
    • 第二种方法
  • 步骤
  • • 一、在mysql创建数据库表
    • 二、开发代码

需求

日常生活中，当我们打开地图时，会通过地图道路颜色获取当前交通情况，也可以通过地图上经常网购的IP地址热力图得出哪些地区网购观念更发达，还有当前疫情的情况，各个地区疫情的热力图可以直观反应出疫情的严重程度。
想要获取热力图，首先要清楚，通过点击流日志中的IP地址信息，可以推算出所在城市甚至经度和维度，通过计算同一经度纬度出现的次数，就可以绘制出热力图

思路

现在拥有两张信息表：
一张是IP字典，这里面可以对我们有直接实用价值的就是有序的转换为Long类型的IP范围，还有每一个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
1.2.8.0|1.2.8.255|16910336|16910591|亚洲|中国|北京|北京||中国互联网信息中心|110100|China|CN|116.405285|39.904989
1.2.9.0|1.2.127.255|16910592|16941055|亚洲|中国|广东|广州||电信|440100|China|CN|113.280637|23.125178
1.3.0.0|1.3.255.255|16973824|17039359|亚洲|中国|广东|广州||电信|440100|China|CN|113.280637|23.125178
1.4.1.0|1.4.3.255|17039616|17040383|亚洲|中国|福建|福州||电信|350100|China|CN|119.306239|26.075302
……
……

另一张是用户IP上网记录，虽然有很多很长，但是对于目前此项目，只需要提取其中的IP地址就可以了

20090121000132095572000|125.213.100.123|show.51|/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 / EmbeddedWB- 14.59  from: / )|.php|
20090121000132124542000|117.101.215.133|www.jiayuan|/19245971|Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; TencentTraveler 4.0)|.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%2F45%2F1b%2F388111afac8195cc5d91ea286cdd%3A1%3A%3Ahttp%3A%2F%2Fimages.love21cn%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%2Fshowphoto.php%3Fuid_hash%3D0319bc5e33ba35755c30a9d88aaf46dc%26total%3D6%26p%3D5; click_count=0%2C3363619
20090121000132406516000|117.101.222.68|gg.xiaonei|/view.jsp?p=389|Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; CIBA)|.do?id=229670724|_r01_=1; __utma=204579609.31669176.1231940225.1232462740.1232467011.145; __utmz=204579609.1231940225.1.1.utmccn=(direct)
20090121000132581311000|115.120.36.98|/tj.htm|Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; TheWorld)|/|
20090121000132864647000|123.197.64.247|cul.sohu|/20071227/n254338813_22.shtml|Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; TheWorld)|.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|/|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/wz/wyfx/it.html; rtime=0; ltime=1232464387281; cnzz_eid=6264952-1232464379-http%3A//www.itxls/wz/wyfx/it.html
20090121000133331104000|123.197.66.93|www.pkwutai|/down/downLoad-id-45383.html|Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; QQDownload 1.7)|=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|

核心思路就是将上网记录中的IP地址转换为Long类型，利用二分法和IP字典中的IP范围比较，找出对于的经度和维度，然后对经度纬度作计数，最后把拿到的数据放到mysql数据库中

ip地址转换为Long类型的两种方法

以下方法是从博主 @徐志/忘川风华录 的文章 IP地址转换成Long型数字算法和原理（全网最细！！）
查到的，佩服的可以点击链接前往观摩

ip地址转换数字地址的原理

IP地址一般是一个32位的二进制数意思就是如果将IP地址转换成二进制表示应该有32为那么长，但是它通常被分割为4个“8位二进制数”（也就是4个字节每，每个代表的就是小于2的8 次方）。IP地址通常用“点分十进制”表示成（d）的形式，其中，a,b,c,d都是0~255之间的十进制整数。例：点分十进IP地址（100.4.5.6），实际上是32位二进制数（01100100.00000100.00000101.00000110）

第一种方法

val ipV4="125.213.100.123"
val fragments = ip22.split("[.]")
var Ip_Num=125*256*256*256+213*256*256+100*256+123
println(Ip_Num)     //打印的结果2111136891

第二种方法

val ip22="125.213.100.123"
val fragments = ip22.split("[.]")
var Ip_Num = 0L
for (i <- 0 until fragments.length){Ip_Num =  fragments(i).toLong | ipNum << 8L
}
println(Ip_Num)     //打印结果  2111136891

步骤

一、在mysql创建数据库表

/*
SQLyog Ultimate v8.32 
MySQL - 5.6.22-log : Database - spark
*********************************************************************
*//*!40101 SET NAMES utf8 */;/*!40101 SET SQL_MODE=''*/;/*!40014 SET @OLD_UNIQUE_CHECKS=@@UNIQUE_CHECKS, UNIQUE_CHECKS=0 */;
/*!40014 SET @OLD_FOREIGN_KEY_CHECKS=@@FOREIGN_KEY_CHECKS, FOREIGN_KEY_CHECKS=0 */;
/*!40101 SET @OLD_SQL_MODE=@@SQL_MODE, SQL_MODE='NO_AUTO_VALUE_ON_ZERO' */;
/*!40111 SET @OLD_SQL_NOTES=@@SQL_NOTES, SQL_NOTES=0 */;
CREATE DATABASE /*!32312 IF NOT EXISTS*/`spark` /*!40100 DEFAULT CHARACTER SET utf8 */;USE `spark`;/*Table structure for table `iplocation` */DROP TABLE IF EXISTS `iplocation`;CREATE TABLE `iplocation` (`longitude` varchar(32) DEFAULT NULL,`latitude` varchar(32) DEFAULT NULL,`total_count` varchar(32) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8;/*!40101 SET SQL_MODE=@OLD_SQL_MODE */;
/*!40014 SET FOREIGN_KEY_CHECKS=@OLD_FOREIGN_KEY_CHECKS */;
/*!40014 SET UNIQUE_CHECKS=@OLD_UNIQUE_CHECKS */;
/*!40111 SET SQL_NOTES=@OLD_SQL_NOTES */;

二、开发代码

import java.sql.{Connection, DriverManager}import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.broadcast.Broadcastobject IpLocation {//定义把IP地址转换为Long类型的方法 125.213.100.123def ipToLong(ip: String): Long = {val split: Array[String] = ip.split("\.")var resultNum: Long = 0Lfor (eachNum <- split) {resultNum = Long | resultNum << 8L}resultNum}/*** 使用二分查找法，确定用户ip落在了哪个范围** @param userLongIP* @param ipArray* @return 返回ipArray的下标索引*/def binarySearch(userLongIP: Long, ipArray: Array[(String, String, String, String)]): Int = {var startIndex: Int = 0var endIndex: Int = ipArray.length//创建while循环while (startIndex <= endIndex) {var middle: Int = {startIndex + endIndex} / 2//判断用户落在哪个范围if (userLongIP >= ipArray(middle)._1.toLong && userLongIP <= ipArray(middle)._2.toLong) {//使用return可以直接返回数据并停掉while循环return middle}//如果落在了middle的左边if (userLongIP < ipArray(middle)._1.toLong) {endIndex = middle - 1}//如果落在了middle的右边if (userLongIP > ipArray(middle)._1.toLong) {startIndex = middle + 1}}//避免方法报错，随便返回一个值即可，因为肯定不会到这个值，while循环最终都会落在return middle上0}/*** 定义函数，将数据放到mysql中去*///datas获取一个分区里的所有数据val data2Mysql = (datas: Iterator[((String, String), Int)]) => {//连接到mysqlval connection: Connection = Connection("jdbc:mysql://localhost:3306/spark", "root", "123456")//使用sql语句导入数据val preparedStatement = connection.prepareStatement("insert into iplocation(longitude,latitude,total_count) values(?,?,?)")//foreach能获取到datas中的每一条数据datas.foreach(record => {preparedStatement.setString(1, record._1._1) //获取经度preparedStatement.setString(2, record._1._2) //获取纬度preparedStatement.setInt(3, record._2) //获取次数ute()})preparedStatement.close()connection.close()}def main(args: Array[String]): Unit = {//获取SparkConfval sparkConf = new SparkConf().setAppName("IP-Location").setMaster("local[2]").set("spark.driver.host", "localhost")//获取SparkContextval sparkContext = new SparkContext(sparkConf)//设置日志筛选条件sparkContext.setLogLevel("WARN")//读取IP字典val IpDictionary: RDD[String] = File("/Users/zhaozhuang/Desktop/4、Spark/Spark第二天/第二天教案/资料/服务器访问日志根据ip地址查找区域/用户ip上网记录以及ip字典/ip.txt")//IP字典内的数据需要的是IP字段Long类型的起始和结束，还有经度和维度//首先需要将拿到的数据进行分割  1.0.1.0|1.0.3.255|16777472|16778239|亚洲|中国|福建|福州||电信|350100|China|CN|119.306239|26.075302val map: RDD[Array[String]] = IpDictionary.map(x => x.split("\|"))//切割后，我们需要的数据就是ip范围起始值、结束值、经度和纬度 下标分别是2,3,length-2,length-1val ipRange: RDD[(String, String, String, String)] = map.map(x => (x(2), x(3), x(x.length - 2), x(x.length - 1)))//将IP字典的数据放到广播变量中，供所有的block块使用val collect: Array[(String, String, String, String)] = llect()val broadcast: Broadcast[Array[(String, String, String, String)]] = sparkContext.broadcast(collect)//读取用户上网记录表val userIpRecord: RDD[String] = File("/Users/zhaozhuang/Desktop/4、Spark/Spark第二天/第二天教案/资料/服务器访问日志根据ip地址查找区域/用户ip上网记录以及ip字典/20090121000132.394251.http.format")//用户上网记录表中只需要IP地址的字段，所以进行切割拿到IP地址字段即可//数据样式 20090121000132095572000|125.213.100.123|show.51|/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 / EmbeddedWB- 14.59  from: / )|.php|val allUserIP: RDD[String] = userIpRecord.map(x => x.split("\|")(1))/*** mapPartitions需要两个参数* def mapPartitions[U: ClassTag](* f: Iterator[T] => Iterator[U],       第一个是给一个是迭代器的函数，返回一个迭代器* preservesPartitioning: Boolean = false)        第二个参数为默认值** 本方法是将拿到的用户IP一次性提取一个分区的数据,iter就表示一个分区的所有数据*/val numTimes: RDD[((String, String), Int)] = allUserIP.mapPartitions(iter => {//获取广播变量里的值   数据样式：16777472|16778239|119.306239|26.075302val ipArray: Array[(String, String, String, String)] = broadcast.value//通过map方法遍历一个分区中的所有数据iter.map(ip => {//获取每一个用户的IP，需要是Long类型val userLongIP: Long = ipToLong(ip)//将用户ip和广播变量中的数据进行比较，确定一下用户ip在哪个ip范围，返回值是ipArray索引val resultIndex: Int = binarySearch(userLongIP, ipArray)//每个经纬度出现一次就计作一次((ipArray(resultIndex)._3, ipArray(resultIndex)._4), 1)})})//将每个经度和维度出现的次数进行累加val locationCount: RDD[((String, String), Int)] = duceByKey(_ + _)//将最终累加的结果值保存到Mysql中去locationCount.foreachPartition(data2Mysql)}
}