map算子：

 object suanziTest {

    def main(args: Array[String]): Unit = {

            val conf = new SparkConf().setAppName("sparkTest").setMaster("local")

                 val sc = new SparkContext(conf)

                  val maprdd: RDD[Int] = sc.makeRDD(1.to(10))

                  //方式一:spark

                  //    val result = maprdd.map(_*10)

                  //    result.foreach(println(_))

                  //    sc.stop()

                  //方式二：scala

                  val ints = maprdd.collect()

                  var result=for( x <- ints) yield (x*10)

                         result.foreach(println(_))

                          }

    }

filter算子：
根据条件返回符合条件的数据，并生成一个新的RDD

val conf = new SparkConf().setAppName("sparkTest").setMaster("local")

val sc = new SparkContext(conf)

val maprdd: RDD[Int] = sc.makeRDD(1.to(10))

val unit = maprdd.filter(_%2==0)

unit.foreach(println(_))

比如有一个集合，元素值是1到10，将偶数拿出来对集合中的数值进行同一乘以10

maprdd.filter(_%2==0).map(_*10).foreach(println(_))

flatMap算子：
将函数体中计算之后的集合对象，打散（压平）

val strings = Array("hello java","hello scala")

val unit: RDD[String] = sc.makeRDD(strings)

//将一行的数据转为一个元素

val unit1: RDD[String] = unit.flatMap(_.split(" "))

//将一行的数据转为一个数组

val unit2: RDD[Array[String]] = unit.map(_.split(" "))

groupByKey算子：
将相同Key的值放在同一个序列中（集合的一种）

val data= List(("za",12),("za",45),("dd",13))

val value: RDD[(String, Int)] = sc.makeRDD(data)

val value1: RDD[(String, Iterable[Int])] = value.groupByKey()

value1.map(x=>{

      var sum:Int=0

      for(v <- x._2){

        sum+=v

      }

      //(x._1,sum)

      x._1+" "+sum

      }).foreach(println(_))

      //foreach(x=>println(x._1+" "+x._2))

      sc.stop()

 }

reduceByKey算子：
将相同的key的值，进行计算之后统一返回

//前面一个下划线表示的是，每一次叠加之后的结果

val data= List(("za",12),("za",45),("dd",13))

val value: RDD[(String, Int)] = sc.makeRDD(data)

value.reduceByKey(_+_).foreach(println(_))

达到的效果和groupByKey一样

union算子：
如果有多个Rdd可以将多个Rdd合并成一个，将后面的rdd的元素，追加到原来的元素，并生成一个新的RDD

var dataA=List(("zs",30),("zs",50),("ls",30))

var dataB=List(("zs",111),("zs",2222),("ls",3333))

//将基础数据转为RDD

val rddA: RDD[(String, Int)] = sc.makeRDD(dataA)

val rddB: RDD[(String, Int)] = sc.makeRDD(dataB)

val value: RDD[(String, Int)] = rddA.union(rddB)

value.foreach(println(_))

join算子：
也是发生在两个rdd之上的。其原理与sql中的inner join一致

将RDDA中的第一个元素拿出来，和RDDB的第一个元素进行匹配，如果KEY相同的话将会组合成一个新的RDD元素（key,(value1,value2)）

如果RDDA中有一个元素在RDDB中没有一个匹配的话，将会出现什么结果？

没有匹配的话将不会显示出来，等于mysql中的inner join方式

var dataA=List(("zs",30),("zs",50),("ls",30),("ww",100))

var dataB=List(("zs",111),("zs",2222),("ls",3333),("zl",3000))

//将基础数据转为RDD

val rddA: RDD[(String, Int)] = sc.makeRDD(dataA)

val rddB: RDD[(String, Int)] = sc.makeRDD(dataB)

val value: RDD[(String, (Int, Int))] = rddA.join(rddB)

value.foreach(println(_))

结果：
(ls,(30,3333))

(zs,(30,111))

(zs,(30,2222))

(zs,(50,111))

(zs,(50,2222))

mapValues算子：
对一个map类型中的value值进行统一操作

var dataA=List(("zs",30),("zs",50),("ls",30),("ww",100))

一、//sc.makeRDD(dataA).map(x=>(x._1,x._2*10)).foreach(println(_))

二、sc.makeRDD(dataA).mapValues(_*10).foreach(println(_)) 

partitionBy算子：
如果有自定义分区的需求的话，可以采用该方式进行处理

如果只需要改变分区的数量的话，有没有必要做自定义分区？

可以采用repartiton(3)算子来进行处理，3为分区数

如何做自定义分区?
1、创建自定义分区类
import org.apache.spark.Partitioner
　　class MyPartition(val numPartition:Int) extends Partitioner{
　　//定义有多少个分区
　　override def numPartitions = {
　　numPartition;
　　}

　　//定义分区的规则
override def getPartition(key: Any) = {
　　val values: String = key.toString
　　if(values.startsWith("135")){
　　　　0
　　}else{
　　　　1
　　}
　　}
}

2、object suanziTest {
　　def main(args: Array[String]): Unit = {
　　　　val conf = new SparkConf().setAppName("sparkTest").setMaster("local")
　　　　val sc = new SparkContext(conf)
　　　　var dataA=List(("zs",30),("zs",50),("ls",30),("ww",100))
　　　　//原来分区
　　　　println( sc.makeRDD(dataA).getNumPartitions)
　　　　//使用自定义分区
　　　　val partiton = new MyPartiton(2)
　　　　println(sc.makeRDD(dataA).partitionBy(partiton).getNumPartitions)
　　　　sc.makeRDD(dataA).partitionBy(partiton).foreach(println(_))
　　　　sc.stop()
　　}
}

count：
统计RDD中有多少个元素

println(sc.makeRDD(dataA).count())

collect:将RDD转为scala中的数组

val value: RDD[Int] = sc.makeRDD(dataA)

val ints: Array[Int] = value.collect()

注：有时候在传递参数的时候，人家要求要一个数组，而自己手里面只有一个RDD的时候，就可以采用这种方式【相当于一种特殊的类型转换】

reduce:
与scala中reduce一样，要求的格式不是一个key、value结构

对于用于reduce算子的，只能有值，spark中的reduce没有类型上的要求

val data = 1.to(10)

val unit = sc.makeRDD(data)

println(unit.reduce(_ + _))

lookup:
要求的RDD类型必须是一个key、value类型

val strings = List("za","ds","fd")

val unit = sc.makeRDD(strings)

//将元素转换为元组,并从RDD中找到key为za的元素

val ints = unit.map((_,1)).lookup("za")

ints.foreach(println(_))

或者：unit.map((_,1)).foreach(x=>{

          if(x._1.equals("za")){

              println(x._2)

            }

          })

或者 val va: RDD[(String, Int)] = unit.map((_,1)).filter(_._1.equals("za"))

     va.map((_._2)).foreach(println(_))

问题：lookup和直接使用foreach的区别是什么？

foreach的方式：拿出每一个元素，通过if条件的方式进行比较，适合条件的进行输出，
而lookup针对于多个分区的时候，会先将需要查询的值（如"za"先进行分区计算--可以定位za具体在哪一个分区中）那么查询该值的时候，就只需要从该分区中拿到值。
相同之处：如果一个RDD只有一个分区的时候，那么foreach等于lookup的操作，如果多条件查询的话，lookup会需要进行多次的分区操作，而foreach只需要进行一次

sortBy:排序

val tuples = sc.makeRDD(Array(("cc",12),("bb",32),("cc",22),("aa",18),("bb",16),("dd",16),("ee",54),("cc",1),("ff",13),("gg",32),("bb",4)))

// 统计key出现的次数

val counts = tuples.reduceByKey(_+_)

// 按照value进行降序排序

val sorts = counts.sortBy(_._2,false).foreach(println(_))

val unit = 1.to(10)

sc.makeRDD(unit).sortBy(x=>x,false).foreach(println(_))

take算子:
取出rdd中的前三个元素

sc.makeRDD(dataA).sortBy(x=>x,false).take(3).foreach(println(_)) take(3)表示取出RDD的前三个元素 sortBy(x=>x,false)排序

first算子：
等于take(1)   拿出rdd的第一个元素

tuples.first()

saveAsTextFile：
将结果输出到指定的目录中

val unit = 1.to(10)

sc.makeRDD(unit).saveAsTextFile("d:/out")

saveAsSequenceFile（序列化文件）：
将结果输出到指定的目录中，而且文件的类型为SequenceFile    要求为    RDD的元素必须由key-value对组成

sc.makeRDD(unit).map((_,1)).saveAsSequenceFile("D:/OUT")  map((_,1))转为key-value形式