Spark之所以比Hadoop灵活和强大，其中一个原因是Spark内置了许多有用的算子，也就是方法。通过对这些方法的组合，编程人员就可以写出自己想要的功能。说白了spark编程就是对spark算子的使用，下面为大家详细讲解一下SparkValue类型的常用算子

map() 接收一个函数，该函数将RDD中的元素逐条进行映射转换，可以是类型的转换，也可以是值的转换，将函数的返回结果作为结果RDD编程。

defmap[U:ClassTag](f:T=>U):RDD[U]案例演示

valsparkConf=newSparkConf().setMaster("local[*]").setAppName("Operator")valsc=newSparkContext(sparkConf)//算子-mapvalrdd=sc.makeRDD(List(1,2,3,4),2)valmapRdd1=rdd.map(_*2)mapRdd1.collect().foreach(println)sc.stop()

运行结果

2468mapPartitons函数说明：

将待处理的数据以分区为单位发送到待计算节点上进行处理，mapPartition是对RDD的每一个分区的迭代器进行操作，返回的是迭代器。这里的处理可以进行任意的处理。

defmapPartitions[U:ClassTag](f:Iterator[T]=>Iterator[U],preservesPartitioning:Boolean=false):RDD[U]案例演示

defmain(args:Array[String]):Unit={valsparkConf=newSparkConf().setMaster("local[*]").setAppName("Operator")valsc=newSparkContext(sparkConf)//算子-mapPartitons计算每个分区的最大数valrdd=sc.makeRDD(List(1,34,36,345,2435,2342,62,35,4),4)valmapParRdd=rdd.mapPartitions(iter=>{List(iter.max).iterator})mapParRdd.foreach(println)sc.stop()}

运行结果：

62243534345mapPartitonsWithIndex函数说明：

将待处理的数据以分区为单位发送到计算节点上，这里的处理可以进行任意的处理，哪怕是过滤数据，在处理的同时可以获取当前分区的索引值。

defmapPartitionsWithIndex[U:ClassTag](f:(Int,Iterator[T])=>Iterator[U],preservesPartitioning:Boolean=false):RDD[U]案例演示：

defmain(args:Array[String]):Unit={valconf=newSparkConf().setMaster("local[*]").setAppName("rdd")valsc=newSparkContext(conf)valrdd=sc.makeRDD(List("HelloSpark","HelloScala","WordCount"),2)valmapRDD=rdd.flatMap(_.split(""))valmpwiRdd=mapRDD.mapPartitionsWithIndex((index,datas)=>{datas.map(num=>{(index,num)})})mpwiRdd.collect().foreach(println)}

运行结果：

(0,Hello)(0,Spark)(1,Hello)(1,Scala)(1,Word)(1,Count)

defmain(args:Array[String]):Unit={valconf=newSparkConf().setMaster("local[*]").setAppName("rdd")valsc=newSparkContext(conf)valrdd=sc.makeRDD(List("HelloSpark","HelloScala","WordCount"),2)valmapRDD=rdd.flatMap(_.split(""))valmpwiRdd=mapRDD.mapPartitionsWithIndex((index,datas)=>{if(index==0){datas.map(num=>{(index,num)})}else{Nil.iterator}})mpwiRdd.collect().foreach(println)

运行结果：

(0,Hello)(0,Spark)flatMap函数说明：

将数据进行扁平化之后在做映射处理，所以算子也称为扁平化映射

defflatMap[U:ClassTag](f:T=>TraversableOnce[U]):RDD[U]案例演示：

将每个单词进行扁平化映射

defmain(args:Array[String]):Unit={valsparkConf=newSparkConf().setMaster("local[*]").setAppName("Operator")valsc=newSparkContext(sparkConf)//算子-mapvalrdd=sc.makeRDD(List("HelloScala","HelloSpark"),2)valFltRdd=rdd.flatMap(_.split(""))FltRdd.foreach(println)sc.stop()}

运行结果：

HelloScalaHelloSparkglom函数说明：

glom的作用就是将一个分区的数据合并到一个array中。

defglom():RDD[Array[T]]案例演示：

defmain(args:Array[String]):Unit={valconf=newSparkConf().setMaster("local[*]").setAppName("rdd")valsc=newSparkContext(conf)valrdd=sc.makeRDD(List(1,2,3,4,5,6,7,8,9),2)valglomRdd=rdd.glom()glomRdd.collect().foreach(data=>println(data.mkString(",")))sc.stop()}

运行结果：

1,2,3,45,6,7,8,9groupBy函数说明：

将数据根据指定的规则进行分组，分区默认不变，单数数据会被打乱，我们成这样的操作为shuffer，

defgroupBy[K](f:T=>K)(implicitkt:ClassTag[K]):RDD[(K,Iterable[T])]案例演示：

defmain(args:Array[String]):Unit={valconf=newSparkConf().setMaster("local[*]").setAppName("rdd")valsc=newSparkContext(conf)valrdd=sc.makeRDD(List(1,2,3,4,5,6,7,8,10),2)valgroupByRDD=rdd.groupBy(_%2==0)groupByRDD.collect().foreach(println)sc.stop()}

运行结果：

(false,CompactBuffer(1,3,5,7))(true,CompactBuffer(2,4,6,8,10))

defmain(args:Array[String]):Unit={valconf=newSparkConf().setMaster("local[*]").setAppName("rdd")valsc=newSparkContext(conf)valrdd=sc.makeRDD(List("Hello","Tom","Timi","Scala","Spark"))valgroupByRDD=rdd.groupBy(_.charAt(0))groupByRDD.collect().foreach(println)sc.stop()}

运行结果：

(T,CompactBuffer(Tom,Timi))(H,CompactBuffer(Hello))(S,CompactBuffer(Scala,Spark))filter函数说明：

filter即过滤器的意思，所以filter算子的作用就是过滤的作用。filter将根据指定的规则进行筛选过滤，符合条件的数据保留，不符合的数据丢弃，当数据进行筛选过滤之后，分区不变，但分区内的数据可能不均衡，生产环境下，可能会出现数据倾斜。

deffilter(f:T=>Boolean):RDD[T]案例演示：

defmain(args:Array[String]):Unit={valconf=newSparkConf().setMaster("local[*]").setAppName("rdd")valsc=newSparkContext(conf)valrdd=sc.makeRDD(List(46,235,246,2346,3276,235,234,6234,6245,246,24,6246,235,26,265))valfilterRDD=rdd.filter(_%2==0)filterRDD.collect().foreach(println)sc.stop()}

运行结果：

4624623463276234623424624624626

2.筛选单词中包含H的

defmain(args:Array[String]):Unit={valconf=newSparkConf().setMaster("local[*]").setAppName("rdd")valsc=newSparkContext(conf)valrdd=sc.makeRDD(List("Hello","Horber","Hbeer","ersfgH","Scala","Hadoop","Zookeeper"))valfilterRDD=rdd.filter(_.contains("H"))filterRDD.collect().foreach(println)sc.stop()}

运行结果：

HelloHorberHbeerersfgHHadoop