一、RDD的创建

1、由一个已经存在的Scala集合创建

2、由外部存储系统的文件创建

  包括本地的文件系统，还有所有Hadoop支持的数据集，比如HDFS、Cassandra、HBase等。

3、已有的RDD经过算子转换生成新的RDD

三、RDD编程API

1.RDD 的算子分类

• Transformation（转换）：根据数据集创建一个新的数据集，计算后返回一个新RDD；例如：一个rdd进行map操作后生了一个新的rdd。
• Action（动作）：对rdd 结果计算后返回一个数值value 给驱动程序；例如：collect 算子将数据集的所有元素收集完成返回给驱动程序。

2.Transformation

  RDD 中的所有转换都是延迟加载的，也就是说，它们并不会直接计算结果。相反的，它们只是记住这些应用到基础数据集（例如一个文件）上的转换动作。只有当发生一个要求返回结果给Driver的动作时，这些转换才会真正运行。这种设计让Spark 更加有效率地运行。

转换	含义
map(func)	返回一个新的 RDD，该 RDD 由每一个输入元素经过 func 函数转换后组成
filter(func)	返回一个新的 RDD，该 RDD 由经过 func 函数计算后返回值为 true 的输入元素组成
flatMap(func)	类似于 map，但是每一个输入元素可以被映射为 0 或多个输出元素（所以 func 应该返回一个序列，而不是单一元素）
mapPartitions(func)	类似于 map，但独立地在 RDD 的每一个分片上运行，因此在类型为 T 的 RDD 上运行时，func 的函数类型必须是 Iterator[T] => Iterator[U]
mapPartitionsWithIndex(func)	类似于 mapPartitions，但 func 带有一个整数参数表示分片的索引值，因此在类型为 T 的 RDD 上运行时，func 的函数类型必须是 (Int, Interator[T]) => Iterator[U]
union(otherDataset)	对源 RDD 和参数 RDD 求并集后返回一个新的 RDD
intersection(otherDataset)	对源 RDD 和参数 RDD 求交集后返回一个新的 RDD
distinct([numTasks]))	对源 RDD 进行去重后返回一个新的 RDD
groupByKey([numTasks])	在一个(K,V)的 RDD 上调用，返回一个(K, Iterator[V])的 RDD
reduceByKey(func, [numTasks])	在一个(K,V)的 RDD 上调用，返回一个(K,V)的 RDD，使用指定的 reduce 函数，将相同 key 的值聚合到一起，与 groupByKey 类似，reduce 任务的个数可以通过第二个可选的参数来设置
sortByKey([ascending], [numTasks])	在一个(K,V)的 RDD 上调用，K 必须实现 Ordered 接口，返回一个按照 key 进行排序的(K,V)的 RDD
sortBy(func,[ascending], [numTasks])	与 sortByKey 类似，但是更灵活
join(otherDataset, [numTasks])	在类型为(K,V)和(K,W)的 RDD 上调用，返回一个相同 key 对应的所有元素对在一起的(K,(V,W))的 RDD
cogroup(otherDataset, [numTasks])	在 类 型 为 (K,V) 和 (K,W) 的 RDD 上 调 用 ， 返 回 一 个 (K,(Iterable<V>,Iterable<W>))类型的 RDD
coalesce(numPartitions)	减少 RDD 的分区数到指定值。
repartition(numPartitions)	重新给 RDD 分区
repartitionAndSortWithinPartitions(part itioner)	重新给 RDD 分区，并且每个分区内以记录的 key 排序

3.Action

动作	含义
reduce(func)	reduce 将 RDD 中元素前两个传给输入函数，产生一个新的 return 值， 新产生的 return 值与 RDD 中下一个元素（第三个元素）组成两个元素，再被传给输入函数，直到最后只有一个值为止。
collect()	在驱动程序中，以数组的形式返回数据集的所有元素
count()	返回 RDD 的元素个数
first()	返回 RDD 的第一个元素（类似于 take(1)）
take(n)	返回一个由数据集的前 n 个元素组成的数组
takeOrdered(n, [ordering])	返回自然顺序或者自定义顺序的前 n 个元素
saveAsTextFile(path)	将数据集的元素以 textfile 的形式保存到 HDFS 文件系统或者其他支持的文件系统，对于每个元素，Spark 将会调用 toString 方法，将它装换为文件中的文本
saveAsSequenceFile(path)	将数据集中的元素以 Hadoop sequencefile 的格式保存到指定的目录下，可以使 HDFS 或者其他 Hadoop 支持的文件系统。
saveAsObjectFile(path)	将数据集的元素，以 Java 序列化的方式保存到指定的目录下
countByKey()	针对(K,V)类型的 RDD，返回一个(K,Int)的 map，表示每一个 key 对应的元素个数。
foreach(func)	在数据集的每一个元素上，运行函数func
foreachPartition(func)	在数据集的每一个分区上，运行函数func

四、RDD 常用的算子操作

  启动 spark-shell进行测试：

    spark-shell--master spark://node1:7077

练习1：map、filter
  map：对每一个元素进行操作
  filter：对每一个元素进行过滤

    //通过并行化生成rdd
    val rdd1 = sc.parallelize(List(5, 6, 4, 7, 3, 8, 2, 9, 1, 10))
    //对rdd1 里的每一个元素乘 2 然后排序
    val rdd2 = rdd1.map(_ * 2).sortBy(x => x, true)
    //过滤出大于等于 5 的元素
    val rdd3 = rdd2.filter(_ >= 5)
    //将元素以数组的方式在客户端显示
    rdd3.collect

练习 2：flatMap

    val rdd1 = sc.parallelize(Array("a b c", "d e f", "h i j"))
    //将rdd1 里面的每一个元素先切分在压平
    val rdd2 = rdd1.flatMap(_.split(" ")) 
    rdd2.collect

练习 3：交集、并集

val rdd1 = sc.parallelize(List(5, 6, 4, 3))
val rdd2 = sc.parallelize(List(1, 2, 3, 4))
//求并集
val rdd3 = rdd1.union(rdd2)
//求交集
val rdd4 = rdd1.intersection(rdd2)
//去重
rdd3.distinct.collect 
rdd4.collect

练习 4：join、groupByKey
  join：将集合中相同的键提取出来，对应的值组成一个元组
  join：将集合中相同的键提取出来，对应的值封装到compactBuffer中

val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 3),("kitty", 2)))
val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1),("shuke", 2)))
//求 join
val rdd3 = rdd1.join(rdd2) 
rdd3.collect
//求并集
val rdd4 = rdd1 union rdd2 
rdd4.collect
//按key 进行分组
val rdd5=rdd4.groupByKey 
rdd5.collect

join groupByKey

练习 5：cogroup

val rdd1 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2)))
val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("jim", 2)))
//cogroup
val rdd3 = rdd1.cogroup(rdd2)
//注意cogroup 与groupByKey的区别
rdd3.collect

cogroup

练习 6：reduce

val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5))
//reduce 聚合
val rdd2 = rdd1.reduce(_ + _) 
rdd2.collect

练习 7：reduceByKey、sortByKey

val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 3),("kitty", 2),("shuke", 1)))
val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 3),("shuke", 2), ("kitty", 5))) val rdd3 = rdd1.union(rdd2)
//按key 进行聚合
val rdd4 = rdd3.reduceByKey(_ + _) 
rdd4.collect
//按 value 的降序排序
val rdd5 = rdd4.map(t => (t._2, t._1)).sortByKey(false).map(t=> (t._2, t._1)) 
rdd5.collect

练习 8：repartition、coalesce

val rdd1 = sc.parallelize(1 to 10,3)
//利用repartition 改变rdd1 分区数
//减少分区
rdd1.repartition(2).partitions.size
//增加分区
rdd1.repartition(4).partitions.size
//利用coalesce 改变rdd1 分区数
//减少分区
rdd1.coalesce(2).partitions.size

  注意：repartition可以增加和减少rdd中的分区数，coalesce只能减少 rdd分区数，增加rdd分区数不会生效。