前言

filter，可以理解为过滤，直观来说，就是对一组数据按照指定的规则做过滤，filter这个算子在Java或者其他语言中多有使用，能够很方便的帮我们从一组数据中过滤得到期望的数据；

函数签名

def filter(f: T => Boolean ): RDD[T]

函数说明

将数据根据指定的规则进行筛选过滤，符合规则的数据保留，不符合规则的数据丢弃。 当数据进行筛选过滤后，分区不变，但是分区内的数据可能不均衡，生产环境下，可能会出现 数据倾斜；

案例一，从一组数据中过滤出偶数

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Filter_Test {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)

    val rdd  = sc.makeRDD(List(1,2,3,4,5,6))
    val result = rdd.filter(
      item => item % 2 ==0
    )
    result.collect().foreach(println)


    sc.stop()

  }

}

运行这段代码，观察控制台输出结果

 

案例二，从日志文件中过滤出2015年5月17的数据

日志文件内容如下： 

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Filter_Test {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)


    val rdd: RDD[String] = sc.textFile("E:\\code-self\\spi\\datas\\apache.log")
    rdd.filter(
      line =>{
        val datas = line.split(" ")
        val time = datas(3)
        time.contains("17/05/2015")
      }
    ).collect().foreach(println)

    sc.stop()

  }

}

运行上面的代码，观察控制台输出结果，