RDD Cache

  1. import org.apache.spark.rdd.RDD
  3. val rootPath: String = _
  4. val file: String = s"${rootPath}/wikiOfSpark.txt"
  6. // 读取文件内容
  7. val lineRDD: RDD[String] = spark.sparkContext.textFile(file)
  9. // 以行为单位做分词
  10. val wordRDD: RDD[String] = lineRDD.flatMap(line => line.split(" "))
  11. val cleanWordRDD: RDD[String] = wordRDD.filter(word => !word.equals(""))
  13. // 把RDD元素转换为（Key，Value）的形式
  14. val kvRDD: RDD[(String, Int)] = cleanWordRDD.map(word => (word, 1))
  16. // 按照单词做分组计数
  17. val wordCounts: RDD[(String, Int)] = kvRDD.reduceByKey((x, y) => x + y)
  19. wordCounts.cache// 使用cache算子告知Spark对wordCounts加缓存
  20. wordCounts.count// 触发wordCounts的计算，并将wordCounts缓存到内存
  22. // 打印词频最高的5个词汇
  23. wordCounts.map{case (k, v) => (v, k)}.sortByKey(false).take(5)
  25. // 将分组计数结果落盘到文件
  26. val targetPath: String = _
  27. wordCounts.saveAsTextFile(targetPath)
  28. take和saveAsTextFile这两个操作执行的都很慢。
  29. cache函数并不会立即触发RDD在内存中的物化，因此还需要调用count算子来触发这一执行过程。
  30. cache函数实际上会进一步调用persisit来为其添加缓存，因此.cache与.persist(MEMORY_ONLY)是等价的

https://time.geekbang.org/column/article/418079
https://time.geekbang.org/column/article/421566
https://time.geekbang.org/column/article/423131

常用算子图表

image.png


Transformations

map

给定映射函数f，map(f)以元素为粒度对RDD做数据转换。其中f可以是带有明确签名的带名函数，也可以是匿名函数，它的型参类型必须与RDD的元素类型保持一致

  1. // 把普通RDD转换为Paired RDD
  3. // 匿名函数
  4. val cleanWordRDD: RDD[String] = _ // 省略完整代码
  5. val kvRDD: RDD[(String, Int)] = cleanWordRDD.map(word => (word, 1))
  7. // 带名函数
  8. def f(word: String): (String, Int) = {
  9.     return (word, 1)
  10. }
  11. val kvRDD: RDD[(String, Int)] = cleanWordRDD.map(f)

map的性能考虑

  2. // 把普通RDD转换为Paired RDD
  4. import java.security.MessageDigest
  6. val cleanWordRDD: RDD[String] = _ // 请参考第一讲获取完整代码
  8. val kvRDD: RDD[(String, Int)] = cleanWordRDD.map{ word =>
  9.   // 获取MD5对象实例
  10.   val md5 = MessageDigest.getInstance("MD5")
  11.   // 使用MD5计算哈希值
  12.   val hash = md5.digest(word.getBytes).mkString
  13.   // 返回哈希值与数字1的Pair
  14.   (hash, 1)
  15. }

对于RDD中的每一条数据记录，我们都需要实力化一个MessageDigest对象来计算这个元素的哈希值。如果RDD有上百万甚至上亿级别的数据记录，实力化对象的开销就会聚成沙。因此，可以用mapPartitions和mapPartitionsWithIndex来解决类似的问题

mapPartitions

以数据分区为粒度的数据转换

  2. // 把普通RDD转换为Paired RDD
  4. import java.security.MessageDigest
  6. val cleanWordRDD: RDD[String] = _ // 请参考第一讲获取完整代码
  8. val kvRDD: RDD[(String, Int)] = cleanWordRDD.mapPartitions( partition => {
  9.   // 注意！这里是以数据分区为粒度，获取MD5对象实例
  10.   val md5 = MessageDigest.getInstance("MD5")
  11.   val newPartition = partition.map( word => {
  12.   // 在处理每一条数据记录的时候，可以复用同一个Partition内的MD5对象
  13.     (md5.digest(word.getBytes()).mkString,1)
  14.   })
  15.   newPartition
  16. })

image.png

flatMap

从元素到集合，再从集合到元素
image.png

  2. // 读取文件内容
  3. val lineRDD: RDD[String] = _ // 请参考第一讲获取完整代码
  4. // 以行为单位提取相邻单词
  5. val wordPairRDD: RDD[String] = lineRDD.flatMap( line => {
  6.   // 将行转换为单词数组
  7.   val words: Array[String] = line.split(" ")
  8.   // 将单个单词数组，转换为相邻单词数组
  9.   for (i <- 0 until words.length - 1) yield words(i) + "-" + words(i+1)
  10. })

image.png

filter

  2. // 定义特殊字符列表
  3. val list: List[String] = List("&", "|", "#", "^", "@")
  5. // 定义判定函数f
  6. def f(s: String): Boolean = {
  7. val words: Array[String] = s.split("-")
  8. val b1: Boolean = list.contains(words(0))
  9. val b2: Boolean = list.contains(words(1))
  10. return !b1 && !b2 // 返回不在特殊字符列表中的词汇对
  11. }
  13. // 使用filter(f)对RDD进行过滤
  14. val cleanedPairRDD: RDD[String] = wordPairRDD.filter(f)

Actions

1.将分布式计算结果直接落盘的操作，如DataFrame的write, RDD的saveAsTextFile
2.将分布式结果收集到Driver端的操作，如first, take, collect