说明
类似 scala 集合中的 filter()
根据规则 对数据进行筛选过滤，符合规则的数据保留，不符合规则的数据丢弃。
例子
val sparConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparConf)
val rdd:RDD[String] = sc.makeRDD( List("李雷","韩梅梅","lili","lucy"),2 )
val filterRDD: RDD[String] = rdd.filter(_.take(1)=="l")
filterRDD.foreach(println)
sc.stop()
分区内数据不变
当数据进行筛选过滤后，分区内的数据不变，即：过滤后，属于 0分区 的数据仍然在 0分区，不会跑到 1分区
验证
val sparConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
val sc = new SparkContext(sparConf)
//设置2个分区
val rdd:RDD[String] = sc.makeRDD( List("李雷","韩梅梅","lili","lucy"),2 )
//过滤前保存到文件中
rdd.saveAsTextFile("res")
val rdd2: RDD[String] = rdd.filter(_.take(1)=="l")
//过滤后保存到文件中
rdd2.saveAsTextFile("res2")
sc.stop()
执行结果：
如下图，过滤后，不符合条件的数据移除了，符合条件的数据仍在原分区中，不会改变所在分区。
数据倾斜
过滤后，分区内的数据可能不均衡，生产环境下，可能会出现 数据倾斜，即：过滤后，0分区 的数据可能还剩1个，1分区 的数据还剩100个
案例
从访问日志中，筛选出 9 点的 数据
数据：
49.74.3.207 - - [14/Jun/2021:00:01:12 +0800] "GET /show_1EF4h7aLPSFB.html HTTP/1.1" 200 17300
121.226.148.220 - - [14/Jun/2021:00:01:14 +0800] "GET /show_1C7cGrpMhQ2.html HTTP/1.1" 200 21979
117.95.183.206 - - [14/Jun/2021:00:02:41 +0800] "GET /show_1EF1cWTFezUR.html HTTP/1.1" 200 15432
157.55.39.186 - - [14/Jun/2021:00:03:58 +0800] "GET /show_1EF5fipw0O0n.html HTTP/1.1" 200 20482
203.208.60.42 - - [14/Jun/2021:00:04:12 +0800] "GET /show_1IX51xrrA8z.html HTTP/1.1" 200 16986
203.208.60.29 - - [14/Jun/2021:00:04:14 +0800] "POST /ajaxApi/addViewNum HTTP/1.1" 200 40
203.208.60.114 - - [14/Jun/2021:00:04:48 +0800] "GET /show_1EF3uYu8CMSP.html HTTP/1.1" 200 20777
49.87.13.200 - - [14/Jun/2021:00:06:03 +0800] "GET /show_1EF2Pg2sje3N.html HTTP/1.1" 200 29320
49.87.13.200 - - [14/Jun/2021:00:06:04 +0800] "GET /ghost/css/bootstrap.min.css HTTP/1.1" 200 117305
49.87.13.200 - - [14/Jun/2021:00:06:04 +0800] "GET /ghost/css/font-awesome.min.css HTTP/1.1" 200 23739
49.87.13.200 - - [14/Jun/2021:00:06:04 +0800] "GET /ghost/css/magnific-popup.min.css HTTP/1.1" 200 5913
49.87.13.200 - - [14/Jun/2021:00:06:04 +0800] "GET /ghost/css/screen.css?v=1 HTTP/1.1" 200 17496
代码
val sparConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparConf)
val rdd:RDD[String] = sc.textFile("C:\\Users\\yumin\\Desktop\\localhost_access_log.2021-06-14.txt")
//val filterRDD: RDD[String] = rdd.map(_.split(" ")(3).slice(13,15))
val filterRDD: RDD[String] = rdd.filter(_.split(" ")(3).slice(13,15)=="09")
filterRDD.foreach(println)
sc.stop()