SparkStream实战之黑名单过滤

小咪咪 2023-06-15 12:27 41阅读 0赞

package com.zgw.spark
    
    import org.apache.log4j.{ Level, Logger}
    import org.apache.spark.SparkConf
    import org.apache.spark.rdd.RDD
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    import org.apache.spark.streaming.dstream.{ DStream, ReceiverInputDStream}
    
    /** * Created by Zhaogw&Lss on 2019/11/22. * 黑名单过滤 * 20180808，zs * 20180808，ls * 20180808，ww * 黑名单列表 * zs * ls */
    object TransFormApp { 
      def main(args: Array[String]): Unit = { 
        val sc: SparkConf = new SparkConf().setMaster("local[3]").setAppName("NetWork").set("spark.testing.memory", "2147480000")
    
        Logger.getLogger("org").setLevel(Level.ERROR)
        //创建StreamingContext两个参数 SparkConf和batch interval
        val ssc = new StreamingContext(sc, Seconds(5))
        /** * 构建黑名单==>(zs,true)(ls，true) */
          val blacks = List("zs","ls")
          val BlakesRdd: RDD[(String, Boolean)] = ssc.sparkContext.parallelize(blacks).map(x=>(x,true))
    
    
    
        //20180808，zs===》(zs:20180808，zs)
        val textStream: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop000", 9999)
        val clickLog: DStream[String] = textStream.map(x => (x.split(",")(1), x)).transform(rdd => { 
          rdd.leftOuterJoin(BlakesRdd).filter(x => x._2._2.getOrElse(false) != true).map({ 
            x => x._2._1
          })
        })
    
       clickLog.print()
    
        ssc.start()
    
        ssc.awaitTermination()
      }
    }

**`linux`中键入下列代码**

[hadoop@hadoop000 ~]$ nc -lk 9999
    20180808,zs
    20180808,ls
    20180808,ww

**`console`打印过滤的信息**

-------------------------------------------
    Time: 1574408000000 ms
    -------------------------------------------
    
    -------------------------------------------
    Time: 1574408005000 ms
    -------------------------------------------
    
    -------------------------------------------
    Time: 1574408010000 ms
    -------------------------------------------
    20180808,ww
    
    -------------------------------------------
    Time: 1574408015000 ms
    -------------------------------------------

### 需求：实现一个黑名单过滤的`Spark Stream`程序，比如黑名单中有`zs，ls`两名用户，那么他们的信息将会被过滤掉 ###

##### 数据的转化规则 #####

**输入日志(`Dstream`)**

20180808,zs
    20180808,ls
    20180808,ww

**黑名单列表(`RDD`)**

zs
    ls

**最终输出**

20180808,ww

### 实现思路： ###

##### 1. 使用 `val blacks = List("zs","ls")`模拟黑名单列表，真实场景数据应该存在`Mysql`中 #####

##### 2. 构建黑名单，将数组转化为`KV`形式`(zs:true),(ls:true),(ww:true)` 并转化为`RDD` #####

`val BlakesRdd: RDD[(String, Boolean)] = ssc.sparkContext.parallelize(blacks).map(x=>(x,true))`

##### 3 先将数据流转化为`KV`形式，输入`20180808,zs`先根据逗号进行分割,把第一位名字取出转化为 `(zs:20180808，zs),(ls:20180808，ls),(ww:20180808，ww)` #####

##### 4 将DStream和Rdd关联[transform][]函数 #####

##### 5 将数据流和黑名单做左连接 #####

（zs:<20180808，zs>,  <true> ）
    （ls:<20180808，ls>,  <true> ）
    （ww:<20180808，ww>,  <false> ）

并过滤出第二位是`true`的数据，并取出`tuple1`中的数据

**核心代码**

textStream.map(x => (x.split(",")(1), x)).transform(rdd => { 
          rdd.leftOuterJoin(BlakesRdd).filter(x => x._2._2.getOrElse(false) != true).map({ 
            x => x._2._1

代码托管于[https://github.com/daizikaikou/learningSpark][https_github.com_daizikaikou_learningSpark]

[transform]: http://spark.apache.org/docs/latest/streaming-programming-guide.html#transformations-on-dstreams
[https_github.com_daizikaikou_learningSpark]: https://github.com/daizikaikou/learningSpark