RDD编程练习

RDD

RDD创建

1、使用sc.textFile(“文件的路径”)从文件系统中加载,sc是SparkContext

2、通过并行集合创建

val array = Array(1,2,3,4,5)
val rdd = sc.parallelize(array)//sc是SparkContext

RDD操作转换得到的RDD是 惰性操作 ，也就是说， 整个转换( transformation)过程只是记录了转换的轨迹，并不会发生真正的计算，只有遇到动作（Action）操作时，才会发生真正的计算，开始从血缘关系源头开始，进行物理的转换操作

常用的转换操作( transformation)

• filter(func) 筛选出满足函数func的元素，并返回一个新的数据集

• map(func) 将每个元素传递到函数func中，并将结果返回为一个新的数据集 一对一

• flatMap(func) 与map()相似，但每个输入元素都可以映射到0或多个输出结果 可以一对多
• groupByKey() 应用于(K,V)键值对的数据集时，返回一个新的(K, Iterable)形式的数据集
• reduceByKey(func) 应用于(K,V)键值对的数据集时，返回一个新的(K, V)形式的数据集，其中每个值是将每个key传递到函数func中进行聚合后的结果

常用的动作操作（Action）

• count() 返回数据集中的元素个数
• collect() 以数组的形式返回数据集中的所有元素
• first() 返回数据集中的第一个元素
• take(n) 以数组的形式返回数据集中的前n个元素
• reduce(func) 通过函数func（输入两个参数并返回一个值）聚合数据集中的元素
• foreach(func) 将数据集中的每个元素传递到函数func中运行

RDD持久化RDD采用惰性求值的机制， 每次遇到行动操作，都会从头开始执行计算 。每次调用行动操作，都会触发一次从头开始的计算。这对于迭代计算而言，代价是很大的

persist()方法可以 标记 为持久化

• persist(MEMORY_ONLY)：表示将RDD作为反序列化的对象存储于JVM中，如果内存不足，就要按照LRU原则替换缓存中的内容
• persist(MEMORY_AND_DISK)表示将RDD作为反序列化的对象存储在JVM中，如果内存不足，超出的分区将会被存放在硬盘上
  一般而言，使用cache()方法时，会调用persist(MEMORY_ONLY)
  可以使用unpersist()方法手动地把持久化的RDD从缓存中移除

RDD分区

为什么要分区 ： 增加并行度，减少通信开销

RDD分区的一个原则是使得分区的个数尽量等于集群中的CPU核心（core）数目

对于不同的Spark部署模式而言（本地模式、Standalone模式、YARN模式、Mesos模式），都可以通过设置 spark.default.parallelism 这个参数的值，来配置默认的分区数目

一般而言：

• 本地模式：默认为本地机器的CPU数目，若设置了local[N],则默认为N
• Apache Mesos：默认的分区数为8
• Standalone或YARN：在“集群中所有CPU核心数目总和”和“2”二者中取较大值作为默认值

手动创建分区

sc.textFile("文件的路径",2) //2就是分区数
sc.parallelize(array,2)  //2就是分区数
repartition(1) //此方法可以重新定义分区 

Pair RDD

键值对RDD 主要使用map()函数来实现

常用的对键值对RDD的转换操作

reduceByKey(func)

groupByKey()

keys

values

sortByKey()默认升序 sortByKey(false)降序

mapValues(func)

join 输入(K,V1)和(K,V2)输出 (K,(V1,V2))

combineByKey

共享变量广播变量 Broadcast Variables

scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))
broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0)

scala> broadcastVar.value
res0: Array[Int] = Array(1, 2, 3)

累加器Accumulators

scala> val accum = sc.longAccumulator("My Accumulator")
accum: org.apache.spark.util.LongAccumulator = LongAccumulator(id: 0, name: Some(My Accumulator), value: 0)

scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))
...
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s

scala> accum.value
res2: Long = 10

数据的读写本地文件系统的数据读写

//读文件
scala> val  textFile = sc. textFile("file:///usr/local/spark/mycode/wordcount/word.txt")
//写文件
scala> textFile. saveAsTextFile("file:///usr/local/spark/mycode/wordcount/writeback")

读取hdfs上的文件

scala> val  textFile = sc.textFile("hdfs://localhost:9000/user/hadoop/word.txt")
scala> textFile.first()
//或者
scala> val textFile = sc.textFile("/user/hadoop/word.txt")
scala> val textFile = sc.textFile("word.txt")
//写文件
scala> textFile.saveAsTextFile("writeback")

读取json文件 并解析

val  jsonStr = sc.textFile("file:///usr/local/spark/examples/src/main/resources/people.json")

使用JSON.parseFull(jsonString:String)解析

部署到集群进入到spark 安装目录

bin/spark-submit \
  --class <main-class> \
  --master <master-url> \
  --deploy-mode <deploy-mode> \
  --conf <key>=<value> \
  ... # other options
  <application-jar> \
  [application-arguments]

• –class 入口类名（例如org.apache.spark.examples.SparkPi）
• –master 群集的主URL（例如spark://23.195.26.187:7077）
• –deploy-mode 是在工作节点（cluster）上部署驱动程序还是在本地部署外部客户端（client）（默认值: client)
• –conf：key = value格式的任意Spark配置属性。对于包含空格的值，在引号中包含“key = value”
• application-jar：捆绑jar的路径，包括您的应用程序和所有依赖项。URL必须在群集内部全局可见，例如，所有节点上都存在的hdfs://路径或file://路径。
• application-arguments：main 方法的传入参数