Spark基础二之RDD编程
RDD弹性分布式数据集
Python、Java、Scala中任意类型的对象或者用户自定义的对象，形成的不可变的对象集合。每个RDD可以被分成多个分区运行在集群中不同节点之上。所谓弹性是指当某个节点挂掉，Spark可以利用重算特性算出丢掉的分区数据。
一、创建RDD
（1）读取外部数据，每一行就是一个集合元素
// 返回的是一个RDD对象
val inputRDD = sc.textFile("/opt/spark/test/log.txt")
（2）程序中对集合进行并行化
val inputRDD = sc.parallelize(List("java","scala","python"))
二、RDD操作
转化操作：返回一个新的RDD类型，如map()、filter()
行动操作：触发实际计算，返回其他类型的结果Arrar[String]，如count()、first()、collect()
//惰性计算：只有在一次行动操作中用到时，才会执行之前的转化操作。譬如textFile就不会一开是加载整个日志文件，而是明白了整个行动目的后，才会加载文件到内存，而且读到有10行error日志即停止。若当某个RDD需要反复被用到，那么可以使用rdd.persist()强制该RDD完整加载到内存中
val errorRDD = inputRDD.filter(line => line.contains("error"))
errorRDD.take(10).foreach(println)
//向Spark传递函数：大部分转化操作与一部分行动操作需要依赖用户传递的函数来计算。当出现NotSerializableException，通常是因为我们传递了一个不可序列化的类中的函数或字段。还有当传递的对象包含某个超大对象的引用，spark会把整个超大对象发布到工作节点而导致内存不够失败
scala> def containsErrors(s:String) = s.contains("error")
containsErrors: (s: String)Boolean
scala> inputRDD.filter(containsErrors)
res8: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[10] at filter at <console>:28
常见转化操作：
2019032701.png
常见行动操作：
2019032702.png
RDD类型转化：
有些函数只能用于特定类型的 RDD，比如 mean() 和 variance() 只能用在数值 RDD 上， 而 join() 只能用在键值对 RDD 上。Scala中通过隐式转换可以把 RDD[Double] 转为DoubleRDDFunctions，然后调用 mean()
持久化：
Scala和 Java中，默认情况下persist() 会把数据以序列化的形式缓存在 JVM 的堆空间中。当然也可以自定义缓存级别。RDD还有一个方法叫作unpersist()，调用该方法可以手动把持久化的 RDD 从缓存中移除
2019032703.png
val result = input.map(x => x * x)
// 缓存，避免了两次行动操作导致加载两次RDD
result.persist(StorageLevel.DISK_ONLY)
println(result.count())
println(result.collect().mkString(","))
网友评论