spark过滤器

//过滤你不需要的字段
 rdd.filter(x=>{//过滤器
        val arr =x.split(",")
        val jg =arr(3)
        if(jg.eq("")){//如果为空那就不要了 
          false
        }else{
          true
        }
    })

map进行分组

map(x=>{//按照同类型分组
      val arr =x.split(",")
      val tp =arr(2)
      val jg =arr(3).toFloat
      (tp,jg)

转hasmap，在rdd后面跟一个 }).collectAsMap()即可

    }).groupByKey().map(x=>{
        val tp =x._1 //group分组
        var sum =0.0 //求和
        for (y<- x._2){
            sum+=y//累加同类型的价格
        }
      val age =sum/x._2.size//计算均值
      (tp,age)
    }).collectAsMap()//把元组转成hasmap，value就是map值
     arr(3)=map.get(tp).toString//取出map中对应的值

arr.mkString(",")//把这个数组用字符串逗号链接然后返回

求商品平均值填空案例

1,AAA,饮料,3.5,武汉
2,BBB,饮料,2.5,武汉
3,CCC,烟酒,3.5,上海
11,OOO,烟酒,5.75,上海
4,DDD,饮料,12.5,上海
5,EEE,饮料,22.5,武汉
6,FFF,烟酒,7.5,上海
11,OOO,饮料,8.5,武汉
7,GGG,饮料,4.5,温州
8,HHH,烟酒,9.5,贵阳
9,JJJ,烟酒,2.5,南宁
10,KKK,饮料,5.5,南宁

package spark

import org.apache.spark.sql.SparkSession

object 均值填充 {
  def main(args: Array[String]): Unit = {
    val sparkSession =SparkSession.builder().master("local")
      .appName("均值填充").getOrCreate()
    val sc =sparkSession.sparkContext
    val rdd =sc.textFile("src/product.txt")
    //1.计算各个类别的均值
    val map =rdd.filter(x=>{//过滤器
        val arr =x.split(",")
        val jg =arr(3)
        if(jg.equals("")){//如果为空那就不要了
          false
        }else{
          true
        }
    }).map(x=>{//按照同类型分组
      val arr =x.split(",")
      val tp =arr(2)
      val jg =arr(3).toFloat
      (tp,jg)
    }).groupByKey().map(x=>{
        val tp =x._1 //group分组
        var sum =0.0 //求和
        for (y<- x._2){
            sum+=y//累加同类型的价格
        }
      val age =sum/x._2.size//计算均值
      (tp,age)
    }).collectAsMap()//把元组转成hasmap，value就是map值
    //完成空值填充
    rdd.map(x=>{
      val arr=x.split(",")
      val age =arr(3)
      val tp =arr(2)//取出商品类型
      if (age.equals("")){
          arr(3)=map.get(tp).get.toString//取出map中对应的值
      }
      arr.mkString(",")//把这个数组用字符串逗号链接然后返回
    }).saveAsTextFile("data/out1")
    sparkSession.stop()

  }

}