spark 开发考题！面试题！ 根据IP地址查询归属地，统计归属地IP地址数

spark开发考题！面试题！ 网络开发式运营人才选拔！

题目：

现有一批IP地址(详见文件ip.txt)，需要根据IP地址库信息(详见文件iplib.txt)，查询归属地信息，并统计每一个归属地IP地址的总数。
请编写spark任务并在测试环境上提交运行。统计结果请以文本文件格式请保存在xxxx目录下，源代码请保存在xxxx目录下。
运行结果格式示例：
江苏  移动 241
江西  电信 67
河南  联通 89 

IPLib_test格式：

1.1.1.0 1.1.1.255 广东 电信
2.0.0.0 2.1.1.255 天津 电信

......

IPs_test 格式：

101.100.24.56
144.0.1.100

......

package com.dt.spark.chinatelecomExam;


public class IpUtil {
    public static boolean ipExistsInRange(String ip, String ipSection) {

        ipSection = ipSection.trim();

        ip = ip.trim();

        int idx = ipSection.indexOf('-');

        String beginIP = ipSection.substring(0, idx);

        String endIP = ipSection.substring(idx + 1);

        return getIp2long(beginIP) <= getIp2long(ip) && getIp2long(ip) <= getIp2long(endIP);

    }


    public static long getIp2long(String ip) {

        ip = ip.trim();

        String[] ips = ip.split("\\.");

        long ip2long = 0L;

        for (int i = 0; i < 4; ++i) {

            ip2long = ip2long << 8 | Integer.parseInt(ips[i]);

        }

        return ip2long;

    }


}

package com.dt.spark.Exam

import scala.collection.immutable.HashSet
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

import util.control.Breaks._
import scala.collection.mutable

/**
  *  
  */
object IPQueryForTest {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)
    var masterUrl = "local[8]"
    if (args.length > 0) {
      masterUrl = args(0)
    }
    val sparkConf = new SparkConf().setMaster(masterUrl).setAppName("IPQueryForTest")
    val spark = SparkSession
      .builder()
      .config(sparkConf)
      .getOrCreate()
    val sc = spark.sparkContext
    //数据存放的目录；
    var dataPath = "data/Exam/"
    /**
      * 读取数据，用什么方式读取数据呢？在这里是使用RDD!
      */
    val IPLibRDD: RDD[String] = sc.textFile(dataPath + "IPLib_test.txt")
    val IPsRDD: RDD[String] = sc.textFile(dataPath + "IPs_test.txt")

    /**
      * 在Spark中如何实现mapjoin呢，显然是要借助于Broadcast，会把数据广播到Executor级别让该Executor上的所有任务共享
      * 该唯一的数据，而不是每次运行Task的时候都要发送一份数据的拷贝，这显著的降低了网络数据的传输和JVM内存的消耗
      */
    val IPLibSet: HashSet[String] = HashSet() ++ IPLibRDD.collect()
    val IPLibSetBroadcast: Broadcast[HashSet[String]] = sc.broadcast(IPLibSet)

    println("纯粹通过RDD的方式实现IP地址统计分析:")
    val resultIPQuery: RDD[(String, Long)] = IPsRDD.map(line => {

      var ip = line.trim
      var exists = false
      var locationKey: String = "NoMatchIPQuery"
      var IPValue: Long = 0L
      val iplibSets: HashSet[String] = IPLibSetBroadcast.value
      for (iplib <- iplibSets) {
        val ipSection: String = iplib.split("\t")(0).trim + "-" + iplib.split("\t")(1).trim
        breakable {
          exists = IpUtil.ipExistsInRange(ip, ipSection)
          if (exists == true) {
            locationKey = iplib.split("\t")(2).trim + "\t" + iplib.split("\t")(3).trim
            IPValue = 1L
            break()
          }
        }
      }
      (locationKey, IPValue)
    })

    val result = resultIPQuery.filter(!_._1.contains("NoMatchIPQuery")).reduceByKey(_ + _)
    result.map(line =>
      line._1 + "\t" + line._2).saveAsTextFile(dataPath + "IPQueryResult.txt")
  }
}