spark filter

//Filter fields you don't needrdd.filter(x=>{//filterval arr =x.split(",")val jg =arr(3)if(jg.eq("")){//If it is empty then don'tfalse}else{true}})

map for grouping

map(x=>{//Group by the same typeval arr =x.split(",")val tp = arr(2)val jg =arr(3).toFloat(tp,jg)

Turn to hasmap, followed by a }).collectAsMap() after rdd

 }).groupByKey().map(x=>{val tp =x._1 //group groupingvar sum =0.0 //summationfor (y<- x._2){sum+=y//accumulate prices of the same type}val age =sum/x._2.size//Calculate the mean(tp,age)}).collectAsMap()//Convert the tuple to hasmap, the value is the map valuearr(3)=map.get(tp).toString//Remove the corresponding value in the map

arr.mkString(",")//Link this array with string commas and return

Fill in the blank case for product average value

1,AAA,Beverage,3.5,Wuhan2, BBB, drinks, 2.5, Wuhan3, CCC, Tobacco and Alcohol, 3.5, Shanghai11, OOO, Tobacco and Alcohol, 5.75, Shanghai4, DDD, drinks, 12.5, Shanghai5, EEE, drinks, 22.5, Wuhan6, FFF, Tobacco and Alcohol, 7.5, Shanghai11, OOO, drinks, 8.5, Wuhan7, GGG, drinks, 4.5, Wenzhou8, HHH, Tobacco and Alcohol, 9.5, Guiyang9, JJJ, Tobacco and Alcohol, 2.5, Nanning10, KKK, drinks, 5.5, Nanning

package sparkimport org.apache.spark.sql.SparkSessionobject mean fill {def main(args: Array[String]): Unit = {val sparkSession =SparkSession.builder().master("local").appName("Mean Fill").getOrCreate()val sc = sparkSession.sparkContextval rdd =sc.textFile("src/product.txt")//1. Calculate the mean of each categoryval map =rdd.filter(x=>{//filterval arr =x.split(",")val jg =arr(3)if(jg.equals("")){//If it is empty then don'tfalse}else{true}}).map(x=>{//Group by the same typeval arr =x.split(",")val tp = arr(2)val jg =arr(3).toFloat(tp,jg)}).groupByKey().map(x=>{val tp =x._1 //group groupingvar sum =0.0 //summationfor (y<- x._2){sum+=y//accumulate prices of the same type}val age =sum/x._2.size//Calculate the mean(tp,age)}).collectAsMap()//Convert the tuple to hasmap, the value is the map value//complete empty value fillingrdd.map(x=>{val arr=x.split(",")val age =arr(3)val tp =arr(2)//Remove item typeif (age.equals("")){arr(3)=map.get(tp).get.toString//Remove the corresponding value in the map}arr.mkString(",")//Link this array with string commas and return}).saveAsTextFile("data/out1")sparkSession.stop()}}