Scala--Spark将某网站的爬虫记录进行整理

短命女 2022-04-02 07:24 189阅读 0赞

设计思路：

1.将ip截取出来（多种方式）用map组成二元组（注意区分map，flatmap）将相同ip出现次数统计出来分析得出爬虫ip

2.将多次重复ip且访问密集的设为访问黑名单

3.将同一时间访问某网站的ip整理出来

package Test1225
    
    
    import org.apache.spark.SparkConf
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    
    object Spider01{
    //  练习1 将ip出现次数多的爬虫ip挑选出来
      def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local[*]").setAppName("ip")
        val ssc = new StreamingContext(conf,Seconds(10))
        ssc.sparkContext.setLogLevel("ERROR")
        ssc.checkpoint("d://123/eq")
        val ds = ssc.socketTextStream("lion",9999)
        val paris = ds.map(_.split(" ")(0)).map(x => (x,1)).reduceByKey(_+_).filter(x => x._2 > 5)
    
        val res = paris.updateStateByKey(updateFunction).repartition(1)
        res.saveAsTextFiles("D://124//spider01")//查询到的结果输出到本地目录
        res.print()
        ssc.start()
        ssc.awaitTermination()
        ssc.stop()
      }
    
      def updateFunction(newValues:Seq[Int],runningCount:Option[Int]): Option[Int] ={
    
        val newCounts = newValues.sum
        val sum = runningCount.getOrElse(0)
        Some(newCounts+sum)
      }
    }

package Test1225
    
    import org.apache.spark.SparkConf
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    
    object spider02 {
    //  根据上一题的结果创建一个黑名单hrdd然后禁止加入黑名单的hrdd访问	，将ip不是114.112.141.6的留下
      def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("ip").setMaster("local[*]")
        val ssc =new StreamingContext(conf,Seconds(10))
        ssc.sparkContext.setLogLevel("ERROR")
        val iplist = List(("114.112.141.6",1))
        val hrdd = ssc.sparkContext.parallelize(iplist)
        val ds = ssc.socketTextStream("lion",9999)
        ds.map(_.split(" ")(0)).map(x => (x,1)).transform{rdd => rdd.leftOuterJoin(hrdd).filter(x => if(x._2._2 == None)true else false)}.print(100)
    
        ssc.start()
        ssc.awaitTermination()
    
    
      }
    
    }

package Test1225
    
    import org.apache.spark.SparkConf
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    
    object spider03 {
      //  统计某一时间段内网站访问量
      def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("ip").setMaster("local[*]")
        val ssc = new StreamingContext(conf, Seconds(10))
        ssc.sparkContext.setLogLevel("ERROR")
        //val iplist = List(("114.112.141.6", 1))
        //val hrdd = ssc.sparkContext.parallelize(iplist)
        val ds = ssc.socketTextStream("lion", 9999)
        ds.map(_.split(" ")(0)).map(x => (x, 1)).reduceByKey(_+_).count().print(100)
    
        ssc.start()
        ssc.awaitTermination()
      }
    }