Spark Core案例实操（七）

Leefs 2021-11-03 PM 923℃ 0条

[TOC]

### 一、需求

> 根据访问日志的ip地址做如下操作：
>
> + 计算出访问者的归属地
> + 按照省份，计算出访问次数
> + 将计算好的结果输出到控制台

### 二、数据分析

+ `access.log`日志文件

![17.Spark Core案例实操（七）01.jpg](https://lilinchao.com/usr/uploads/2021/11/2354706471.jpg)

+ 第一列：ID
+ 第二列：访问者IP
+ 第三列：访问网址

后面没用到就不详细介绍了

+ `ip.txt`IP规则文件

![17.Spark Core案例实操（七）02.jpg](https://lilinchao.com/usr/uploads/2021/11/1486800510.jpg)

+ 第一列和第二列：开始IP和结束IP(一个范围)
+ 第三列和第四列：开始IP和结束IP十进制
+ 第五、六、七、八列：对应地区分别是洲、国家、省/直辖市、市/区
+ 第九列：运营商名称

### 三、实现

#### 3.1 步骤

```
1.整理数据，切分出ip字段，然后将ip地址转换成十进制
2.加载规则，整理规则，取出有用的字段，然后将数据缓存到内存中
3.将访问log与ip规则进行匹配（二分法查找）
4.取出对应的省份名称，然后将其和一组合在一起
5.按省份名进行聚合
6.将聚合后的数据在控制台打印
```

#### 3.2 代码实现

```scala
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * @author lilinchao
  * @date 2021/11/3
  * @description 1.0 
  **/
object IpLoaction {

def main(args: Array[String]): Unit = {

val conf = new SparkConf().setAppName("IpLoaction").setMaster("local[*]")
    val sc = new SparkContext(conf)

//获取全部IP规则数据
    val rules: Array[(Long, Long, String)] = MyUtils.readRules("datas/ip.txt")

//调用sc上的广播方法
    val broadcastRef: Broadcast[Array[(Long, Long, String)]]  = sc.broadcast(rules)

//创建RDD，读取访问日志
    val accessLines: RDD[String] = sc.textFile("datas/access.log")

val func = (line:String) => {
      //对数据进行切割
      val fields = line.split("[|]")
      //获取IP
      val ip = fields(1)
      //将IP转换成十进制
      val ipNum = MyUtils.ip2Long(ip)
      //获取广播方法中IP规则
      val rulesInExecutor: Array[(Long, Long, String)] = broadcastRef.value
      //查找
      var province = "未知"
      //在IP规则中进行二分查找（IP规则事先是排好序的）
      val index = MyUtils.binarySearch(rulesInExecutor,ipNum)
      if(index != -1){
        province = rulesInExecutor(index)._3
      }
      (province,1)
    }

//整理数据
    val proviceAndOne: RDD[(String, Int)] = accessLines.map(func)

//聚合
    //val sum = (x: Int, y: Int) => x + y
    val reduced: RDD[(String, Int)] = proviceAndOne.reduceByKey(_+_)

//将结果打印
    val r = reduced.collect()

println(r.toBuffer)

sc.stop()
  }
}
```

**运行结果**

```
ArrayBuffer((陕西,1824), (河北,383), (云南,126), (重庆,868), (北京,1535))
```

### 结尾

因为本篇使用的示例数据`access.log`、`ip.txt`文件由于数据量较大将不在下方贴出。

直接在微信公众号【Java和大数据进阶】回复：**sparkdata**，即可获取。

标签: Spark, Spark Core

非特殊说明，本博所有文章均为博主原创。

如若转载，请注明出处：https://www.lilinchao.com/archives/1607.html

上一篇 Spark Core案例实操（六）

下一篇 Spark Core案例实操（八）

Spark Core案例实操（七）

评论啦~

栏目分类

标签云

友情链接申请

Spark Core案例实操（七）

 评论啦~

 栏目分类

标签云

友情链接申请

评论啦~

栏目分类

标签云

友情链接申请