map(func) 返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成
//map val rdd: RDD[Int] = sc.parallelize(1 to 10) rdd.foreach(println) //将每个元素扩大5倍 val result: RDD[Int] = rdd.map(_*5) //打印输出 result.foreach(println)filter(func) 返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成
//filter var sourceFilter = sc.parallelize(Array("xiaoming","xiaojiang","xiaohe","dazhi")) //过滤出来以da开头的元素 val result: RDD[String] = sourceFilter.filter(_.startsWith("da")) //打印输出 println(result.collect().toBuffer) //过滤出来以e结尾 val result1: RDD[String] = sourceFilter.filter(_.endsWith("e")) println(result1.collect().toBuffer) //过滤出来包含ao的元素 val result2 = sourceFilter.filter(_.contains("ao")) println(result2.collect().toBuffer)flatMap(func) 类似于map，但是每一个输入元素可以被映射为0或多个输出元素（所以func应该返回一个序列，而不是单一元素）
//flatMap val result: RDD[String] = sc.parallelize(Array("a b c", "d e f")) println(result.collect().toBuffer) //切分 val results: RDD[String] = result.flatMap(_.split(" ")) println(results.collect().toBuffer)mapPartitions(func) 类似于map，但独立地在RDD的每一个分片上运行，因此在类型为T的RDD上运行时，func的函数类型必须是Iterator[T] => Iterator[U]
//mapPartitions //将arr转换为RDD val rdd: RDD[Int] = sc.parallelize(Array(1, 2, 3, 4, 5), 1) //针对每个分区当中的元素进行操作 val result: RDD[String] = rdd.mapPartitions(x => Iterator(x.mkString("-"))) //打印输出 println(result.collect().toBuffer)mapPartitionsWithIndex(func) 类似于mapPartitions，但func带有一个整数参数表示分片的索引值，因此在类型为T的RDD上运行时，func的函数类型必须是 (Int, Interator[T]) => Iterator[U]
//mapPartitionsWithIndex val rdd: RDD[Int] = sc.parallelize(1 to 16, 4) //查看每个分区当中都保存了哪些元素 val result: RDD[String] = rdd.mapPartitionsWithIndex((x, y) => Iterator(x + ":" + y.mkString(","))) //打印输出 result.foreach(println)union(otherDataset) 对源RDD和参数RDD求并集后返回一个新的RDD
//union:针对两个RDD当中的元素进行整合，成为一个新的RDD，但是元素不去重 val rdd1: RDD[Int] = sc.parallelize(1 to 6) val rdd2 = sc.parallelize(4 to 8) val result: RDD[Int] = rdd1.union(rdd2) //打印输出 println(result.collect().toBuffer)intersection(otherDataset) 对源RDD和参数RDD求交集后返回一个新的RDD（交集）
//Intersection:求两个RDD的交集，并去重 val rdd1: RDD[Int] = sc.parallelize(1 to 5) val rdd2: RDD[Int] = sc.parallelize(3 to 7) val result: RDD[Int] = rdd1.intersection(rdd2) println(result.collect().toBuffer)distinct([numTasks])) 对源RDD进行去重后返回一个新的RDD
//distinct val lst: List[Int] = List(1, 2, 3, 4, 5, 4, 2, 2, 1) val rdd: RDD[Int] = sc.parallelize(lst) //查看rdd当中的分区 println(rdd.partitions.size) //去重 val rdd1: RDD[Int] = rdd.distinct(3) //打印输出 println(rdd1.collect().toBuffer) //分区数 println(rdd1.partitions.size)groupByKey([numTasks]) 在一个(K,V)的RDD上调用，返回一个(K, Iterator[V])的RDD
//groupByKey val words = Array("one", "two", "two", "three", "three", "three") val rdd: RDD[String] = sc.parallelize(words) val rdd1: RDD[(String, Int)] = rdd.map((_, 1)) val result: RDD[(String, Iterable[Int])] = rdd1.groupByKey() //打印输出 println(result.collect().toBuffer)reduceByKey(func, [numTasks]) 在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，与groupByKey类似，reduce任务的个数可以通过第二个可选的参数来设置
//reduceByKey val rdd = sc.parallelize(List(("female",1),("male",5),("female",5),("male",2))) val result: RDD[(String, Int)] = rdd.reduceByKey(_+_) //打印输出 println(result.collect().toBuffer)aggregateByKey(zeroValue)(seqOp, combOp, [numTasks]) 先按分区聚合 再总的聚合 每次要跟初始值交流 例如：aggregateByKey(0)(+,+) 对k/v的RDD进行操作
//aggregateByKey //第一个参数是初始值，而且参与运算 //第二个参数是分区之内进行操作（小聚合） //第三个参数是分区之间进行操作（大聚合） val rdd = sc.parallelize(List(("a",3),("a",2),("c",4),("b",3),("c",6),("c",8),("a",33),("c",6)),3) //查看每个分区当中都保存了哪些元素 rdd.mapPartitionsWithIndex((x,y)=>Iterator(x+":"+y.mkString(","))).foreach(println) //调用算子 val result: RDD[(String, Int)] = rdd.aggregateByKey(0)(math.max(_, _), _ + _) //查看结果当中每个分区都保存了哪些元素 result.mapPartitionsWithIndex((x,y)=>Iterator(x+":"+y.mkString(","))).foreach(println)sortByKey([ascending], [numTasks]) 在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD
//sortByKey val rdd: RDD[(Int, String)] = sc.parallelize(Array((3, "aa"), (6, "cc"), (2, "bb"), (1, "dd"))) val result: RDD[(Int, String)] = rdd.sortByKey() //打印输出 println(result.collect().toBuffer)sortBy(func,[ascending], [numTasks]) 与sortByKey类似，但是更灵活 第一个参数是根据什么排序 第二个是怎么排序 false倒序 第三个排序后分区数 默认与原RDD一样
//sortBy:默认是升序，如果想降序就false val rdd: RDD[(String, Int)] = sc.parallelize(List(("a", 4), ("c", 2), ("b", 1))) val result: RDD[(String, Int)] = rdd.sortBy(_._2) println(result.collect().toBuffer)join(otherDataset, [numTasks]) 在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD 相当于内连接（求交集）
//join:两个RDD进行join，相同的key进行输出，(k,(v,w))不同的key，不进行输出 val rdd = sc.parallelize(Array((1,"a"),(2,"b"),(3,"c"))) val rdd1 = sc.parallelize(Array((1,4),(2,5),(3,6))) val result: RDD[(Int, (String, Int))] = rdd.join(rdd1) //打印输出 println(result.collect().toBuffer)cogroup(otherDataset, [numTasks]) 在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable,Iterable))类型的RDD
cartesian(otherDataset) 两个RDD的笛卡尔积 的成很多个K/V
//cartesian:两个RDD当中的元素组合，有点多少种组合方式全部输出 val rdd1: RDD[Int] = sc.parallelize(1 to 3) val rdd2: RDD[Int] = sc.parallelize(2 to 7) val result: RDD[(Int, Int)] = rdd1.cartesian(rdd2) //打印输出 println(result.collect().toBuffer)coalesce(numPartitions) 重新分区 第一个参数是要分多少区，第二个参数是否shuffle 默认false 少分区变多分区 true 多分区变少分区 false
//coalesce:默认不产生shuffle val rdd: RDD[Int] = sc.parallelize(1 to 16, 4) //查看每个分区当中保存了哪些元素 rdd.mapPartitionsWithIndex((x,y)=>Iterator(x+":"+y.mkString(","))).foreach(println) //查看分区数 println(rdd.partitions.size) //缩减分区 val rdd2: RDD[Int] = rdd.coalesce(3, true) //查看缩减分区后，每个分区当中保存了哪些元素 rdd2.mapPartitionsWithIndex((x,y)=>Iterator(x+":"+y.mkString(","))).foreach(println) //查看分区数 println(rdd2.partitions.size)repartition(numPartitions) 重新分区 必须shuffle 参数是要分多少区 少变多
//repartition:既可以增加分区，也可以减少分分区 val rdd: RDD[Int] = sc.parallelize(1 to 16, 4) //查看每个分区当中都保存了哪些元素 rdd.mapPartitionsWithIndex((x,y)=>Iterator(x+":"+y.mkString(","))).foreach(println) //查看分区数 println(rdd.partitions.size) //缩减分区数 val rdd2: RDD[Int] = rdd.repartition(3) //查看每个分区当中都保存了哪些元素 rdd2.mapPartitionsWithIndex((x,y)=>Iterator(x+":"+y.mkString(","))).foreach(println) //查看分区数 println(rdd2.partitions.size)foldByKey(zeroValue)(seqOp) 该函数用于K/V做折叠，合并处理 ，与aggregate类似 第一个括号的参数应用于每个V值 第二括号函数是聚合例如：+
combineByKey 合并相同的key的值 rdd1.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
partitionBy（partitioner） 对RDD进行分区 partitioner是分区器 例如new HashPartition(2
//partitionBy:非key，value的元素分区方式为None，key，value的元素可以有分区方式，但是不一定有 val rdd: RDD[(Int, String)] = sc.parallelize(Array((1, "aaa"), (2, "bbb"), (3, "ccc"), (4, "ddd")), 4) //查看分区方式 println(rdd.partitioner) //指定分区方式 val result: RDD[(Int, String)] = rdd.partitionBy(new org.apache.spark.HashPartitioner(3)) //查看分区方式 println(result.partitioner) println(result.partitions.size)cache和persist RDD缓存，可以避免重复计算从而减少时间，区别：cache内部调用了persist算子，cache默认就一个缓存级别MEMORY-ONLY ，而persist则可以选择缓存级别
Subtract（rdd） 返回前rdd元素不在后rdd的rdd（差集）
//subtract:返回前RDD当中的元素不在后RDD的RDD元素 val rdd1: RDD[Int] = sc.parallelize(1 to 5) val rdd2: RDD[Int] = sc.parallelize(3 to 7) val result: RDD[Int] = rdd1.subtract(rdd2) println(result.collect().toBuffer)leftOuterJoin leftOuterJoin类似于SQL中的左外关联left outer join，返回结果以前面的RDD为主，关联不上的记录为空。只能用于两个RDD之间的关联，如果要多个RDD关联，多关联几次即可。
rightOuterJoin rightOuterJoin类似于SQL中的有外关联right outer join，返回结果以参数中的RDD为主，关联不上的记录为空。只能用于两个RDD之间的关联，如果要多个RDD关联，多关联几次即可
subtractByKey substractByKey和基本转换操作中的subtract类似只不过这里是针对K的，返回在主RDD中出现，并且不在otherRDD中出现的元素
//subtract:返回前RDD当中的元素不在后RDD的RDD元素 val rdd1: RDD[Int] = sc.parallelize(1 to 5) val rdd2: RDD[Int] = sc.parallelize(3 to 7) val result: RDD[Int] = rdd1.subtract(rdd2) println(result.collect().toBuffer)