【极简spark教程】RDD编程

入门

val textFile &＃61; sc.textFile("/test.csv")//textFile为RDD类型&＃xff0c;具有List的很多相似操作&＃xff0c;可以进行循环遍历&＃xff0c;例如map,foreach,filter等

1. map操作&＃xff1a;对rdd中每行进行处理
2. flatmap操作&＃xff1a;对rdd中每行进行展开处理
3. collect操作&＃xff1a;将结果转换为Array类型
4. cache操作&＃xff1a;将rdd和dataset保存在内存&＃xff0c;被session持有

RDD编程指引

1. 创建rdd集合&＃xff0c;可以将rdd看做是spark分布式环境下的list

   val data &＃61; Array(1, 2, 3, 4, 5)
val distData &＃61; sc.parallelize(data, 5)// distData类型为ParallelCollectionRDD&＃xff0c;且分片数为5

2. 读取文件
   1. 若读取本地文件&＃xff0c;本地文件需要在所有节点上可以被访问到
   2. 所有读取文件的方法都支持在目录上、通配符、压缩包上运行

      sc.textFile("/my/directory")
sc.textFile("/my/directory/*.txt")
sc.textFile("/my/directory/*.gz")

   3. 控制返回文件数量&＃xff0c;通常情况下返回文件为一个文件夹下的多个文件&＃xff0c;可以使用SparkContext.wholeTextFiles控制返回文件的个数&＃xff0c;例如返回一个文件

      SparkContext.sequenceFile[Int, String]
SparkContext.hadoopRDD
SparkContext.objectFile

3. RDD操作
   1. 转换transform&＃xff1a;生成了新的RDD
      1. map&＃xff1a;返回一个新的分布式数据集&＃xff0c;该数据集是通过将源的每个元素传递给函数func形成的
      2. mapValues&＃xff1a;返回一个新的分布式数据集&＃xff0c;同map相似&＃xff0c;mapValues在(K,V)对的数据集上调用&＃xff0c;仅对V进行操作
      3. filter&＃xff1a;返回一个新的数据集&＃xff0c;该数据集是通过选择源中func返回true的那些元素形成的
      4. flatmap&＃xff1a;与map相似&＃xff0c;但是每个输入项都可以映射到0个或多个输出项&＃xff08;因此func应该返回Seq而不是单个项&＃xff09;
      5. mapPartitions&＃xff1a;与map相似&＃xff0c;但是分别在RDD的每个分区&＃xff08;块&＃xff09;上运行&＃xff0c;因此func在类型T的RDD上运行时必须为Iterator &＃61;> Iterator 类型
      6. mapPartitionsWithIndex&＃xff1a;与mapPartitions相似&＃xff0c;但它还为func提供表示分区索引的整数值&＃xff0c;因此当在类型T的RDD上运行时&＃xff0c;func必须为&＃xff08;Int&＃xff0c;Iterator &＃xff09;&＃61;> Iterator 类型
      7. sample&＃xff1a;使用给定的随机数发生器的种子进行抽样&＃xff0c;共三个参数&＃xff0c;WithReplacement为true表示有抽样放回&＃xff0c;原数据集大小不变&＃xff0c;为false表示无放回抽样&＃xff0c;原数据集在抽样后减少百分比&＃xff0c;fraction表示抽样比例&＃xff0c;seed表示随机数种子&＃xff0c;Long型整数&＃xff0c;例如12345L
      8. union&＃xff1a;返回一个新的数据集&＃xff0c;其中包含源数据集中的元素的并集
      9. intersection&＃xff1a;返回一个新的RDD&＃xff0c;其中包含源数据集中的元素的交集
      10. distinct&＃xff1a;返回一个新的数据集&＃xff0c;其中包含源数据集的不同元素
      11. groupByKey&＃xff1a;在&＃xff08;K&＃xff0c;V&＃xff09;对的数据集上调用时&＃xff0c;返回&＃xff08;K&＃xff0c;Iterable &＃xff09;对的数据集。注意&＃xff1a;如果要分组以便对每个键执行聚合&＃xff08;例如求和或平均值&＃xff09;&＃xff0c;则使用reduceByKey或aggregateByKey将产生更好的性能。注意&＃xff1a;默认情况下&＃xff0c;输出中的并行度取决于父RDD的分区数。您可以传递一个可选numPartitions参数来设置不同数量的任务。
      12. reduceByKey&＃xff1a;在&＃xff08;K&＃xff0c;V&＃xff09;对的数据集上调用时&＃xff0c;返回&＃xff08;K&＃xff0c;V&＃xff09;对的数据集&＃xff0c;其中每个键的值使用给定的reduce函数func&＃xff08;其类型必须为&＃xff08;V&＃xff0c;V&＃xff09;&＃61;>&＃xff09;进行汇总V.与in一样groupByKey&＃xff0c;reduce任务的数量可以通过可选的第二个参数配置
      13. aggregateByKey&＃xff1a;在&＃xff08;K&＃xff0c;V&＃xff09;对的数据集上调用时&＃xff0c;返回&＃xff08;K&＃xff0c;U&＃xff09;对的数据集&＃xff0c;其中每个键的值使用给定的Combine函数和中性的“零”值进行汇总。允许与输入值类型不同的聚合值类型&＃xff0c;同时避免不必要的分配。像in中一样groupByKey&＃xff0c;reduce任务的数量可以通过可选的第二个参数配置
      14. sortByKey&＃xff1a;在由K实现Ordered的&＃xff08;K&＃xff0c;V&＃xff09;对的数据集上调用时&＃xff0c;返回&＃xff08;K&＃xff0c;V&＃xff09;对的数据集&＃xff0c;按布尔值指定&＃xff0c;按键以升序或降序排序ascending
      15. join&＃xff1a;在&＃xff08;K&＃xff0c;V&＃xff09;和&＃xff08;K&＃xff0c;W&＃xff09;类型的数据集上调用时&＃xff0c;返回&＃xff08;K&＃xff0c;&＃xff08;V&＃xff0c;W&＃xff09;&＃xff09;对的数据集&＃xff0c;其中每个键都有所有成对的元素。外连接通过支持leftOuterJoin&＃xff0c;rightOuterJoin和fullOuterJoin。注意&＃xff1a;join之前最好确认rdd中元素的类型&＃xff0c;防止出现Any类型&＃xff0c;导致报错&＃xff1a;but class RDD is invariant in type T.You may wish to define T as &＃43;T instead.
      16. cogroup&＃xff1a;在&＃xff08;K&＃xff0c;V&＃xff09;和&＃xff08;K&＃xff0c;W&＃xff09;类型的数据集上调用时&＃xff0c;返回&＃xff08;K&＃xff0c;&＃xff08;Iterable &＃xff0c;Iterable &＃xff09;&＃xff09;元组的数据集。此操作也称为groupWith
      17. cartesian&＃xff1a;笛卡尔积&＃xff0c;在类型T和U的数据集上调用时&＃xff0c;返回&＃xff08;T&＃xff0c;U&＃xff09;对&＃xff08;所有元素对&＃xff09;的数据集
      18. pipe&＃xff1a;通过shell命令&＃xff08;例如Perl或bash脚本&＃xff09;通过管道传输RDD的每个分区。将RDD元素写入进程的stdin&＃xff0c;并将输出到其stdout的行作为字符串的RDD返回
      19. coalesce&＃xff1a;将RDD中的分区数减少到numPartitions。筛选大型数据集后&＃xff0c;对于更有效地运行操作很有用
      20. repartition&＃xff1a;随机重排RDD中的数据以创建更多或更少的分区&＃xff0c;并在整个分区之间保持平衡。这始终会拖曳网络上的所有数据
          1. repartition(1)&＃xff1a;重排RDD中的数据&＃xff0c;合并为一个分区
          2. repartition(col("colName"))&＃xff1a;重排RDD中的数据&＃xff0c;根据指定列的记录进行分区
      21. repartitionAndSortWithinPartitions&＃xff1a;根据给定的分区程序对RDD重新分区&＃xff0c;并在每个结果分区中&＃xff0c;按其键对记录进行排序。这比repartition在每个分区内调用然后排序更为有效&＃xff0c;因为它可以将排序推入洗牌机制
   2. 行动action&＃xff1a;汇总所有结果返回驱动程序
      1. reduce&＃xff1a;使用函数func&＃xff08;该函数接受两个参数并返回一个&＃xff09;来聚合数据集的元素。该函数应该是可交换的和关联的&＃xff0c;以便可以并行正确地计算它
      2. collect&＃xff1a;在驱动程序中将数据集的所有元素作为数组返回。这通常在返回足够小的数据子集的过滤器或其他操作之后很有用
      3. count&＃xff1a;返回数据集中的元素数
      4. first&＃xff1a;返回数据集的第一个元素&＃xff08;类似于take&＃xff08;1&＃xff09;&＃xff09;
      5. take&＃xff1a;返回数据集的前n个元素的数组
      6. takeSample&＃xff1a;返回一个数组&＃xff0c;该数组包含数据集num个元素的随机样本&＃xff08;是否替换&＃xff09;&＃xff0c;可以选择预先指定随机数生成器种子
      7. takeOrdered&＃xff1a;使用自然顺序或自定义比较器返回RDD 的前n个元素
      8. saveAsTextFile&＃xff1a;将数据集的元素以文本文件&＃xff08;或文本文件集&＃xff09;的形式写入本地文件系统&＃xff0c;HDFS或任何其他Hadoop支持的文件系统中的给定目录中。Spark将在每个元素上调用toString&＃xff0c;以将其转换为文件中的一行文本
      9. saveAsSequenceFile&＃xff1a;在本地文件系统&＃xff0c;HDFS或任何其他Hadoop支持的文件系统的给定路径中&＃xff0c;将数据集的元素作为Hadoop SequenceFile写入。这在实现Hadoop的Writable接口的键/值对的RDD上可用。在Scala中&＃xff0c;它也可用于隐式转换为Writable的类型&＃xff08;Spark包括对基本类型&＃xff08;如Int&＃xff0c;Double&＃xff0c;String等&＃xff09;的转换
      10. saveAsObjectFile&＃xff1a;使用Java序列化以简单的格式编写数据集的元素&＃xff0c;然后可以使用进行加载 SparkContext.objectFile()
      11. countByKey&＃xff1a;仅在类型&＃xff08;K&＃xff0c;V&＃xff09;的RDD上可用。返回&＃xff08;K&＃xff0c;Int&＃xff09;对的哈希图以及每个键的计数
      12. foreach&＃xff1a;在数据集的每个元素上运行函数func。通常这样做是出于副作用&＃xff0c;例如更新累加器或与外部存储系统进行交互。注意&＃xff1a;在之外修改除累加器以外的变量foreach()可能会导致不确定的行为。有关更多详细信息&＃xff0c;请参见了解闭包。
   3. 缓存
      1. persist&＃xff1a;可以根据参数进行不同级别的缓存MEMORY_ONLY、MEMORY_AND_DISK、MEMORY_ONLY_SER、MEMORY_AND_DISK_SER、DISK_ONLY
      2. cache&＃xff1a;默认缓存级别MEMORY_ONLY
      3. 缓存级别选择&＃xff1a;MEMORY_ONLY>MEMORY_ONLY_SER>MEMORY_AND_DISK
      4. unpersist&＃xff1a;释放缓存
   4. 打印部分记录
      1. collect&＃xff1a;将全部记录汇总到一台机器上&＃xff0c;可能会耗尽内存
      2. take&＃xff1a;获取部分记录
   5. 共享变量
      1. 广播变量&＃xff1a;在所有节点上创建一个只读变量&＃xff0c;在使用时不应该调用函数中的指定变量值&＃xff0c;而是直接使用指定广播变量&＃xff0c;而且防止修改节点上的广播变量&＃xff0c;dataFrame和变量都可以使用broadcast进行广播&＃xff0c;但是rdd不可以

         val broadcastVar &＃61; sc.broadcast(Array(1, 2, 3))val broadcastDF &＃61; functions.broadcast(df)

4. 累加器
   1. 创建累加器

      val accum &＃61; sc.longAccumulator("My Accumulator")sc.parallelize(Array(1, 2, 3, 4)).foreach(x &＃61;> accum.add(x))println(accum.value)

   2. 构造累加器

      ​
//继承AccumulatorV2class VectorAccumulatorV2 extends AccumulatorV2[MyVector, MyVector] {
private val myVector: MyVector &＃61; MyVector.createZeroVector
def reset(): Unit &＃61; {myVector.reset()}def add(v: MyVector): Unit &＃61; {myVector.add(v)}
}// 创建累加器对象val myVectorAcc &＃61; new VectorAccumulatorV2
//在spark上下文中进行注册
sc.register(myVectorAcc, "MyVectorAcc1")


   3. 留意惰性(spark2.4.0中疑似取消了&＃xff0c;因为以下代码在spark2.4.0中测试返回了正常结果)

      val accum &＃61; sc.longAccumulator
data.map { x &＃61;> accum.add(x); x }
// 这里累加器仍然为0&＃xff0c;因为没有行动action操作触发执行map操作.