Spark详解(05) - Spark核心编程SparkCore
**Spark详解(05) - Spark核心编程SparkCore
**
RDD概述
RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，是Spark中最基本的数据抽象。
代码中是一个抽象类，它代表一个弹性的、不可变、可分区、里面的元素可并行计算的集合。
- RDD类比工厂生产
- WordCount工作流程
RDD编程
在Spark中创建RDD的创建方式可以分为三种：从集合中创建RDD、从外部存储创建RDD、从其他RDD创建。
从集合中创建
从集合中创建RDD，Spark主要提供了两种函数：parallelize和makeRDD
val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
val rdd: RDD[Int] = sc.parallelize(Array(1, 2, 3, 4, 5, 6, 7, 8))
val rdd1: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4, 5, 6, 7, 8))
注意：makeRDD有两种重构方法，重构方法一如下，makeRDD和parallelize功能一样。
注意：只需要知道makeRDD不完全等于parallelize即可。
def makeRDD[T: ClassTag](seq: Seq[(T, Seq[String])]): RDD[T] = withScope {
val indexToPrefs = seq.zipWithIndex.map(t => (t._2, t._1._2)).toMap
new ParallelCollectionRDD[T](this, seq.map(_._1), math.max(seq.size, 1), indexToPrefs)
从外部存储系统的数据集创建
由外部存储系统的数据集创建RDD包括：本地的文件系统，还有所有Hadoop支持的数据集，比如HDFS、HBase等。
在新建的SparkCoreTest项目名称上右键=》新建input文件夹=》在input文件夹上右键=》分别新建1.txt和2.txt。每个文件里面准备一些word单词。
从其他RDD创建
RDD数据从集合中创建
分区的结束位置 =（分区号 + 1）* 数据总长度/分区总数
RDD数据从文件中读取后创建
RDD整体上分为Value类型、双Value类型和Key-Value类型
Value类型
map()映射
val rdd: RDD[Int] = sc.makeRDD(1 to 4, 2)
val mapRdd: RDD[Int] = rdd.map(_ * 2)
mapRdd.collect().foreach(println)
mapPartitions()以分区为单位执行Map
val rdd: RDD[Int] = sc.makeRDD(1 to 4, 2)
// 2 调用mapPartitions方法，每个元素乘以2
val rdd1 = rdd.mapPartitions(x=>x.map(_*2))
rdd1.collect().foreach(println)
map()和mapPartitions()区别
mapPartitionsWithIndex()带分区号
def mapPartitionsWithIndex[U: ClassTag](
f: (Int, Iterator[T]) => Iterator[U], // Int表示分区编号
preservesPartitioning: Boolean = false): RDD[U]
2）功能说明：类似于mapPartitions，比mapPartitions多一个整数参数表示分区号
3）需求说明：创建一个RDD，使每个元素跟所在分区号形成一个元组，组成一个新的RDD
val rdd: RDD[Int] = sc.makeRDD(1 to 4, 2)
// 2 创建一个RDD，使每个元素跟所在分区号形成一个元组，组成一个新的RDD
val indexRdd = rdd.mapPartitionsWithIndex( (index,items)=>{items.map( (index,_) ¨C817C ¨C818C
indexRdd.collect().foreach(println)
flatMap()扁平化
1）函数签名：def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U]
与map操作类似，将RDD中的每一个元素通过应用f函数依次转换为新的元素，并封装到RDD中。
区别：在flatMap操作中，f函数的返回值是一个集合，并且会将每一个该集合中的元素拆分出来放到新的RDD中。
3）需求说明：创建一个集合，集合里面存储的还是子集合，把所有子集合中数据取出放入到一个大的集合中。
val listRDD=sc.makeRDD(List(List(1,2),List(3,4),List(5¨C845C6¨C846CList¨C847C7¨C848C 2¨C849C
listRDD.flatMap(list=>list).collect.foreach(println)
glom()分区转换数组
1）函数签名：def glom(): RDD[Array[T]]
该操作将RDD中每一个分区变成一个数组，并放置在新的RDD中，数组中元素的类型与原分区中元素类型一致
3）需求说明：创建一个2个分区的RDD，并将每个分区的数据放到一个数组，求出每个分区的最大值
val rdd = sc.makeRDD(1 to 4, 2)
val maxRdd: RDD[Int] = rdd.glom().map(_.max)
groupBy()分组
val rdd = sc.makeRDD(1 to 4, 2)
// 2 将每个分区的数据放到一个数组并收集到Driver端打印
rdd.groupBy(_ % 2).collect().foreach(println)
val rdd1: RDD[String] = sc.makeRDD(List("hello","hive","hadoop","spark","scala"¨C920C
rdd1.groupBy(str=>str.substring(0,1)).collect().foreach(println)
shuffle一定会落盘。可以在local模式下执行程序，通过4040看效果。
GroupBy之WordCount
val strList: List[String] = List("Hello Scala", "Hello Spark", "Hello World")
val wordRdd: RDD[String] = rdd.flatMap(str => str.split(" "))
val wordToOneRdd: RDD[(String, Int)] = wordRdd.map(word => (word, 1))
val groupRdd: RDD[(String, Iterable[(String, Int)])] = wordToOneRdd.groupBy(t => t._1)
filter()过滤
1）函数签名： def filter(f: T => Boolean): RDD[T]
接收一个返回值为布尔类型的函数作为参数。当某个RDD调用filter方法时，会对该RDD中每一个元素应用f函数，如果返回值类型为true，则该元素会被添加到新的RDD中。
val rdd: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4), 2¨C956C
val filterRdd: RDD[Int] = rdd.filter(_ % 2 == 0)
filterRdd.collect().foreach(println)
sample()采样
seed: Long = Utils.random.nextLong): RDD[T]
// withReplacement： true为有放回的抽样，false为无放回的抽样；
// fraction表示：以指定的随机种子随机抽样出数量为fraction的数据；
3）需求说明：创建一个RDD（1-10），从中选择放回和不放回抽样
val dataRDD: RDD[Int] = sc.makeRDD(List(1,2,3,4,5¨C1000C6¨C1001C
伯努利算法：又叫0、1分布。例如扔硬币，要么正面，要么反面。
具体实现：根据种子和随机算法算出一个数和第二个参数设置几率比较，小于第二个参数要，大于不要
第二个参数：抽取的几率，范围在[0,1]之间,0：全不取；1：全取；
val sampleRDD: RDD[Int] = dataRDD.sample(false, 0.5)
sampleRDD.collect().foreach(println)
第一个参数：抽取的数据是否放回，true：放回；false：不放回
第二个参数：重复数据的几率，范围大于等于0.表示每一个元素被期望抽取到的次数
val sampleRDD1: RDD[Int] = dataRDD.sample(true, 2)
sampleRDD1.collect().foreach(println)
distinct()去重
val distinctRdd: RDD[Int] = sc.makeRDD(List(1,2,1,5,2¨C1057C9¨C1058C6¨C1059C1¨C1060C
distinctRdd.distinct().collect().foreach(println)
distinctRdd.distinct(2).collect().foreach(println)
coalesce()合并分区
Coalesce算子包括：配置执行Shuffle和配置不执行Shuffle两种方式。
def coalesce(numPartitions: Int, shuffle: Boolean = false, //默认false不执行shuffle
partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
(implicit ord: Ordering[T] = null) : RDD[T]
2）功能说明：缩减分区数，用于大数据集过滤后，提高小数据集的执行效率。
//val rdd: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4), 4)
//val coalesceRdd: RDD[Int] = rdd.coalesce(2)
val rdd: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4, 5¨C1106C 6¨C1107C 3¨C1108C
val coalesceRDD: RDD[Int] = rdd.coalesce(2)
val indexRDD: RDD[(Int, Int)] = coalesceRDD.mapPartitionsWithIndex(
indexRDD.collect().foreach(println)
//延迟一段时间，观察http://localhost:4040页面，查看Shuffle读写数据
val rdd: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4, 5¨C1167C 6¨C1168C 3¨C1169C
val coalesceRdd: RDD[Int] = rdd.coalesce(2, true)
repartition()重新分区（执行Shuffle）
1）函数签名： def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]
val rdd: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4, 5¨C1226C 6¨C1227C 3¨C1228C
//val coalesceRdd: RDD[Int] = rdd.coalesce(2, true)
val repartitionRdd: RDD[Int] = rdd.repartition(2)
val indexRdd: RDD[(Int, Int)] = repartitionRdd.mapPartitionsWithIndex(
indexRdd.collect().foreach(println)
coalesce和repartition区别
1）coalesce重新分区，可以选择是否进行shuffle过程。由参数shuffle: Boolean = false/true决定。
2）repartition实际上是调用的coalesce，进行shuffle。源码如下：
3）coalesce一般为缩减分区，如果扩大分区，不使用shuffle是没有意义的，repartition扩大分区执行shuffle。
sortBy()排序
ascending: Boolean = true, // 默认为正序排列
numPartitions: Int = this.partitions.length)
(implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]
该操作用于排序数据。在排序之前，可以将数据通过f函数进行处理，之后按照f函数处理的结果进行排序，默认为正序排列。排序后新产生的RDD的分区数与原RDD的分区数一致。
3）需求说明：创建一个RDD，按照数字大小分别实现正序和倒序排序
val rdd: RDD[Int] = sc.makeRDD(List(2, 1, 3, 4, 6¨C1296C 5¨C1297C
val sortRdd: RDD[Int] = rdd.sortBy(num => num)
sortRdd.collect().foreach(println)
val sortRdd2: RDD[Int] = rdd.sortBy(num => num, false)
sortRdd2.collect().foreach(println)
val strRdd: RDD[String] = sc.makeRDD(List("1", "22", "12", "2", "3"¨C1342C
strRdd.sortBy(num => num.toInt).collect().foreach(println)
// 6 先按照tuple的第一个值排序，相等再按照第2个值排
rdd3.sortBy(t=>t).collect().foreach(println)
pipe()调用shell脚本
1）函数签名： def pipe(command: String): RDD[String]
管道，针对每个分区，都调用一次shell脚本，返回输出的RDD。
scala> val rdd = sc.makeRDD (List("hi","Hello","how","are","you"), 1)
scala> rdd.pipe("/opt/module/spark-local/pipe.sh").collect()
res18: Array[String] = Array(Start, >>>hi, >>>Hello, >>>how, >>>are, >>>you)
scala> val rdd = sc.makeRDD(List("hi","Hello","how","are","you"), 2)
scala> rdd.pipe("/opt/module/spark-local/pipe.sh").collect()
res19: Array[String] = Array(Start, >>>hi, >>>Hello, Start, >>>how, >>>are, >>>you)
双Value类型交互
intersection()交集
1）函数签名：def intersection(other: RDD[T]): RDD[T]
val rdd1: RDD[Int] = sc.makeRDD(1 to 4)
val rdd2: RDD[Int] = sc.makeRDD(4 to 8)
rdd1.intersection(rdd2).collect().foreach(println)
union()并集
1）函数签名：def union(other: RDD[T]): RDD[T]
val rdd1: RDD[Int] = sc.makeRDD(1 to 4)
val rdd2: RDD[Int] = sc.makeRDD(4 to 8)
rdd1.union(rdd2).collect().foreach(println)
subtract()差集
1）函数签名：def subtract(other: RDD[T]): RDD[T]
计算差的一种函数，去除两个RDD中相同元素，不同的RDD将保留下来
3）需求说明：创建两个RDD，求第一个RDD与第二个RDD的差集
val rdd: RDD[Int] = sc.makeRDD(1 to 4)
val rdd1: RDD[Int] = sc.makeRDD(4 to 8)
rdd.subtract(rdd1).collect().foreach(println)
zip()拉链
1）函数签名：def zip[U: ClassTag](other: RDD[U]): RDD[(T, U)]
该操作可以将两个RDD中的元素，以键值对的形式进行合并。其中，键值对中的Key为第1个RDD中的元素，Value为第2个RDD中的元素。
将两个RDD组合成Key/Value形式的RDD，这里默认两个RDD的partition数量以及元素数量都相同，否则会抛出异常。
3）需求说明：创建两个RDD，并将两个RDD组合到一起形成一个(k,v)RDD
val rdd1: RDD[Int] = sc.makeRDD(Array(1,2,3),3)
val rdd2: RDD[String] = sc.makeRDD(Array("a","b","c"),3)
rdd1.zip(rdd2).collect().foreach(println)
rdd2.zip(rdd1).collect().foreach(println)
val rdd3: RDD[String] = sc.makeRDD(Array("a","b"), 3)
// Can only zip RDDs with same number of elements in each partition
rdd1.zip(rdd3).collect().foreach(println)
val rdd4: RDD[String] = sc.makeRDD(Array("a","b","c"), 2)
// Can't zip RDDs with unequal numbers of partitions: List(3, 2)
rdd1.zip(rdd4).collect().foreach(println)
Key-Value类型
partitionBy()按照K重新分区
1）函数签名：def partitionBy(partitioner: Partitioner): RDD[(K, V)]
将RDD[K,V]中的K按照指定Partitioner重新进行分区；
如果原有的RDD和新的RDD是一致的话就不进行分区，否则会产生Shuffle过程。
val rdd: RDD[(Int, String)] = sc.makeRDD(Array((1,"aaa"),(2,"bbb"¨C1624C3¨C1625C"ccc"¨C1626C3¨C1627C
val rdd2: RDD[(Int, String)] = rdd.partitionBy(new org.apache.spark.HashPartitioner(2¨C1641C
//3 打印查看对应分区数据 (0,(2,bbb)) (1,(1,aaa)) (1,(3,ccc))
val indexRdd = rdd2.mapPartitionsWithIndex(
(index, datas) => datas.map((index,_))
indexRdd.collect().foreach(println)
自定义分区
class HashPartitioner(partitions: Int) extends Partitioner {
require(partitions >= 0, s"Number of partitions ($partitions) cannot be negative.")
要实现自定义分区器，需要继承org.apache.spark.Partitioner类，并实现下面三个方法。
（1）numPartitions: Int:返回创建出来的分区数。
（2）getPartition(key: Any): Int:返回给定键的分区编号（0到numPartitions-1）。
val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
val rdd: RDD[(Int, String)] = sc.makeRDD(Array((1, "aaa"), (2, "bbb"), (3, "ccc")), 3)
val rdd3: RDD[(Int, String)] = rdd.partitionBy(new MyPartitioner(2))
reduceByKey()按照K聚合V
def reduceByKey(func: (V, V) => V): RDD[(K, V)]
def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]
2）功能说明：该操作可以将RDD[K,V]中的元素按照相同的K对V进行聚合。其存在多种重载形式，还可以设置新RDD的分区数。
val rdd = sc.makeRDD(List(("a",1),("b",5),("a",5),("b",2¨C1689C
val reduce: RDD[(String, Int)] = rdd.reduceByKey((v1,v2) => v1+v2¨C1703C
reduce.collect().foreach(println)
groupByKey()按照K重新分组
1）函数签名：def groupByKey(): RDD[(K, Iterable[V])]
groupByKey对每个key进行操作，但只生成一个seq，并不进行聚合。
该操作可以指定分区器或者分区数（默认使用HashPartitioner）
val rdd = sc.makeRDD(List(("a",1),("b",5),("a",5),("b",2¨C1730C
val group: RDD[(String, Iterable[Int])] = rdd.groupByKey()
group.collect().foreach(println)
group.map(t=>(t._1,t._2.sum)).collect().foreach(println)
reduceByKey和groupByKey区别
1）reduceByKey：按照key进行聚合，在shuffle之前有combine（预聚合）操作，返回结果是RDD[K,V]。
2）groupByKey：按照key进行分组，直接进行shuffle。
3）开发指导：在不影响业务逻辑的前提下，优先选用reduceByKey。求和操作不影响业务逻辑，求平均值影响业务逻辑。
aggregateByKey()按照K处理分区内和分区间逻辑
（1）zeroValue（初始值）：给每一个分区中的每一种key一个初始值；
（2）seqOp（分区内）：函数用于在每一个分区中用初始值逐步迭代value；
（3）combOp（分区间）：函数用于合并每个分区中的结果。
rdd.aggregateByKey(0)(math.max(_, _), _ + _).collect().foreach(println¨C1815C
foldByKey()分区内和分区间相同的aggregateByKey()
1）函数签名： def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]
2）功能说明：aggregateByKey的简化操作，seqop和combop相同。即，分区内逻辑和分区间逻辑相同。
//rdd.aggregateByKey(0)(_+_,_+_).collect().foreach(println)
rdd.foldByKey(0)(_+_).collect().foreach(println)
combineByKey()转换结构后分区内和分区间操作
mergeCombiners: (C, C) => C): RDD[(K, C)]
（2）mergeValue（分区内）: 如果这是一个在处理当前分区之前已经遇到的键，它会使用mergeValue()方法将该键的累加器对应的当前值与这个新的值进行合并
3）需求说明：创建一个pairRDD，根据key计算每种key的均值。（先计算每个key出现的次数以及可以对应值的总和，再相除得到结果）
val input: RDD[(String, Int)] = sc.makeRDD(list, 2)
//2 将相同key对应的值相加，同时记录该key出现的次数，放入一个二元组
val combineRdd: RDD[(String, (Int, Int))] = input.combineByKey(
(acc: (Int, Int), v) => (acc._1 + v, acc¨C1940C_2 ¨C1941C 1¨C1942C
combineRdd.collect().foreach(println)
(key, value._1 / value._2.toDouble)
reduceByKey、foldByKey、aggregateByKey、combineByKey
sortByKey()按照K进行排序
ascending: Boolean = true, // 默认，升序
numPartitions: Int = self.partitions.length) : RDD[(K, V)]
在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD
3）需求说明：创建一个pairRDD，按照key的正序和倒序进行排序
rdd.sortByKey(true).collect().foreach(println)
rdd.sortByKey(false).collect().foreach(println)
mapValues()只对V进行操作
1）函数签名：def mapValues[U](f: V => U): RDD[(K, U)]
3）需求说明：创建一个pairRDD，并将value添加字符串"|||"
rdd.mapValues(_ + "|||").collect().foreach(println)
join()连接
def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]
def join[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (V, W))]
在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD
3）需求说明：创建两个pairRDD，并将key相同的数据聚合到一个元组。
val rdd1: RDD[(Int, Int)] = sc.makeRDD(Array((1, 4), (2¨C2113C 5¨C2114C ¨C2115C4¨C2116C 6¨C2117C
rdd.join(rdd1).collect().foreach(println)
cogroup()类似全连接
1）函数签名：def cogroup[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))]
在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable
操作两个RDD中的KV元素，每个RDD中相同key中的元素分别聚合成一个集合。
3）需求说明：创建两个pairRDD，并将key相同的数据聚合到一个迭代器。
val rdd: RDD[(Int, String)] = sc.makeRDD(Array((1,"a"),(2,"b"¨C2147C3¨C2148C"c"¨C2149C
val rdd1: RDD[(Int, Int)] = sc.makeRDD(Array((1,4),(2,5¨C2163C4¨C2164C6¨C2165C
(1,(CompactBuffer(a),CompactBuffer(4)))
(2,(CompactBuffer(b),CompactBuffer(5)))
(3,(CompactBuffer(c),CompactBuffer()))
(4,(CompactBuffer(),CompactBuffer(6)))
rdd.cogroup(rdd1).collect().foreach(println)
行动算子是触发了整个作业的执行。因为转换算子都是懒加载，并不会立即执行。
reduce()聚合
1）函数签名：def reduce(f: (T, T) => T): T
2）功能说明：f函数聚集RDD中的所有元素，先聚合分区内数据，再聚合分区间数据。
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
val reduceResult: Int = rdd.reduce(_+_)
collect()以数组的形式返回数据集
1）函数签名：def collect(): Array[T]
2）功能说明：在驱动程序中，以数组Array的形式返回数据集的所有元素。
注意：在执行collect()时所有的数据都会被拉取到Driver端，慎用
3）需求说明：创建一个RDD，并将RDD内容收集到Driver端打印
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
rdd.collect().foreach(println)
count()返回RDD中元素个数
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
val countResult: Long = rdd.count()
first()返回RDD中的第一个元素
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
val firstResult: Int = rdd.first()
take()返回由RDD前n个元素组成的数组
1）函数签名：def take(num: Int): Array[T]
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
val takeResult: Array[Int] = rdd.take(2)
println(takeResult.mkString(","))
takeOrdered()返回该RDD排序后前n个元素组成的数组
1）函数签名：def takeOrdered(num: Int)(implicit ord: Ordering[T]): Array[T]
3）需求说明：创建一个RDD，获取该RDD排序后的前2个元素
val rdd: RDD[Int] = sc.makeRDD(List(1,3,2,4))
val result: Array[Int] = rdd.takeOrdered(2)
aggregate()案例
1）函数签名： def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U
2）功能说明：aggregate函数将每个分区里面的元素通过分区内逻辑和初始值进行聚合，然后用分区间逻辑和初始值(zeroValue)进行操作。
注意：分区间逻辑再次使用初始值和aggregateByKey是有区别的。
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 8¨C2365C
//val result: Int = rdd.aggregate(0)(_ + _, _ + _)
val result: Int = rdd.aggregate(10)(_ + _, _ + _)
fold()案例
1）函数签名： def fold(zeroValue: T)(op: (T, T) => T): T
2）功能说明：折叠操作，aggregate的简化操作，即，分区内逻辑和分区间逻辑相同
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
val foldResult: Int = rdd.fold(0)(_+_)
countByKey()统计每种key的个数
1）函数签名：def countByKey(): Map[K, Long]
val result: collection.Map[Int, Long] = rdd.countByKey()
save保存数据相关算子
**1）saveAsTextFile(path)保存成Text文件 **
（2）功能说明：将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本
**2）saveAsSequenceFile(path) 保存成Sequencefile文件 **
（2）功能说明：将数据集中的元素以Hadoop Sequencefile的格式保存到指定的目录下，可以使HDFS或者其他Hadoop支持的文件系统。
**3）saveAsObjectFile(path) 序列化成对象保存到文件 **
（2）功能说明：用于将RDD中的元素序列化成对象，存储到文件中。
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4), 2¨C2479C
rdd.saveAsObjectFile("output1")
rdd.map((_,1)).saveAsSequenceFile("output2")
foreach(f)遍历算子
1）函数签名：def foreach(f: T => Unit): Unit
// val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),2)
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
rdd.collect().foreach(println)
闭包
Spark中的闭包变量一般指，在算子作用域的外部声明，却在算子作用域内操作和执行的变量。
JavaRDD
rdd.foreach(x -> counter += x);
println("Counter value: " + counter);
如果希望在集群模式下，对某个driver端的变量，进行分布式并行地全局性的修改，可以使用Spark提供的Accumulator，全局累加器
序列化方法和属性
Executor：算子里面的代码都是在Executor端执行
val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello spark", "hive", "hadoop"))
def getMatch1 (rdd: RDD[String]): RDD[String] = {
class Search() extends Serializable{…}
def getMatche2(rdd: RDD[String]): RDD[String] = {
rdd.filter(x => x.contains(query))
class Search() extends Serializable{…}
def getMatche2(rdd: RDD[String]): RDD[String] = {
val q = this.query//将类变量赋值给局部变量
rdd.filter(x => x.contains(q))
case class Search(query:String) {…}
Kryo序列化框架
参考地址: https://github.com/EsotericSoftware/kryo
Java的序列化能够序列化任何的类。但是比较重，序列化后对象的体积也比较大。
Spark出于性能的考虑，Spark2.0开始支持另外一种Kryo序列化机制。Kryo速度是Serializable的10倍。当RDD在Shuffle数据的时候，简单数据类型、数组和字符串类型已经在Spark内部使用Kryo来序列化。
注意：即使使用Kryo序列化，也要继承Serializable接口。
object serializable_Kryo {
def main(args: Array[String]): Unit = {
val conf: SparkConf = new SparkConf()
.setAppName("SerDemo")
.setMaster("local[*]")
// 替换默认的序列化机制
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
// 注册需要使用__kryo序列化的自定义类
.registerKryoClasses(Array(classOf[Searche]))
val sc = new SparkContext(conf)
val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello hadoop", " hadoop", "hahah"), 2)
val searche = new Searche("hello")
val result: RDD[String] = searche.getMatchedRDD1(rdd)
result.collect.foreach(println)
}
}
case class Searche(val query: String) {
def isMatch(s: String) = {
s.contains(query)
}
def getMatchedRDD1(rdd: RDD[String]) = {
rdd.filter(isMatch)
}
def getMatchedRDD2(rdd: RDD[String]) = {
val q = query
rdd.filter(_.contains(q))
}
}
查看血缘关系
RDD只支持粗粒度转换，即在大量记录上执行的单个操作。将创建RDD的一系列Lineage（血统）记录下来，以便恢复丢失的分区。RDD的Lineage会记录RDD的元数据信息和转换行为，当该RDD的部分分区数据丢失时，它可以根据这些信息来重新运算和恢复丢失的数据分区。
代码实现
val fileRDD: RDD[String]
= sc.textFile("input/1.txt")
println(fileRDD.toDebugString)
println("----------------------")
val wordRDD: RDD[String]
= fileRDD.flatMap(_.split(" "))
println(wordRDD.toDebugString)
println("----------------------")
val mapRDD: RDD[(String, Int)]
= wordRDD.map((_,1))
println(mapRDD.toDebugString)
println("----------------------")
val resultRDD: RDD[(String, Int)]
= mapRDD.reduceByKey(_+_)
println(resultRDD.toDebugString)
resultRDD.collect()
打印结果
(2) input/1.txt MapPartitionsRDD[1] at textFile at Lineage01.scala:15
[]
| input/1.txt HadoopRDD[0] at textFile at Lineage01.scala:15
[]
----------------------
(2) MapPartitionsRDD[2] at flatMap at Lineage01.scala:19
[]
| input/1.txt MapPartitionsRDD[1] at textFile at Lineage01.scala:15
[]
| input/1.txt HadoopRDD[0] at textFile at Lineage01.scala:15
[]
----------------------
(2) MapPartitionsRDD[3] at map at Lineage01.scala:23
[]
| MapPartitionsRDD[2] at flatMap at Lineage01.scala:19
[]
| input/1.txt MapPartitionsRDD[1] at textFile at Lineage01.scala:15
[]
| input/1.txt HadoopRDD[0] at textFile at Lineage01.scala:15
[]
----------------------
(2) ShuffledRDD[4] at reduceByKey at Lineage01.scala:27
[]
+-(2) MapPartitionsRDD[3] at map at Lineage01.scala:23
[]
| MapPartitionsRDD[2] at flatMap at Lineage01.scala:19
[]
| input/1.txt MapPartitionsRDD[1] at textFile at Lineage01.scala:15
[]
| input/1.txt HadoopRDD[0] at textFile at Lineage01.scala:15
[]
注意：圆括号中的数字表示RDD的并行度，也就是有几个分区
查看依赖关系
1）代码实现
val fileRDD: RDD[String]
= sc.textFile("input/1.txt")
println(fileRDD.dependencies)
println("----------------------")
val wordRDD: RDD[String]
= fileRDD.flatMap(_.split(" "))
println(wordRDD.dependencies)
println("----------------------")
val mapRDD: RDD[(String, Int)]
= wordRDD.map((_,1))
println(mapRDD.dependencies)
println("----------------------")
val resultRDD: RDD[(String, Int)]
= mapRDD.reduceByKey(_+_)
println(resultRDD.dependencies)
resultRDD.collect()
// 查看localhost:4040页面，观察DAG图
Thread.sleep(10000000)
2）打印结果
List(org.apache.spark.OneToOneDependency@f2ce6b)
----------------------
List(org.apache.spark.OneToOneDependency@692fd26)
----------------------
List(org.apache.spark.OneToOneDependency@627d8516)
----------------------
List(org.apache.spark.ShuffleDependency@a518813)
3）全局搜索（ctrl+n）org.apache.spark.OneToOneDependency
class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency[T](rdd) {
override def getParents(partitionId: Int): List[Int] = List(partitionId)
}
注意：要想理解RDDS是如何工作的，最重要的就是理解Transformations。
RDD之间的关系可以从两个维度来理解：一个是RDD是从哪些RDD转换而来，也就是 RDD的parent RDD(s)是什么; 另一个就是RDD依赖于parent RDD(s)的哪些Partition(s)，这种关系就是RDD之间的依赖。
RDD和它依赖的父RDD（s）的关系有两种不同的类型，即窄依赖（narrow dependency）和宽依赖（wide dependency）。
窄依赖
窄依赖表示每一个父RDD的Partition最多被子RDD的一个Partition使用，窄依赖形象的比喻为独生子女。
宽依赖
宽依赖表示同一个父RDD的Partition被多个子RDD的Partition依赖，会引起Shuffle，总结：宽依赖形象的比喻为超生。
具有宽依赖的transformations包括：sort、reduceByKey、groupByKey、join和调用rePartition函数的任何操作。
宽依赖对Spark去评估一个transformations有更加重要的影响，比如对性能的影响。
Stage任务划分（面试重点）
1）DAG有向无环图
DAG（Directed Acyclic Graph）有向无环图是由点和线组成的拓扑图形，该图形具有方向，不会闭环。例如，DAG记录了RDD的转换过程和任务的阶段。
2）任务运行的整体流程
3）RDD任务切分中间分为：Application、Job、Stage和Task
（1）Application：初始化一个SparkContext即生成一个Application；
（2）Job：一个Action算子就会生成一个Job；
（3）Stage：Stage等于宽依赖的个数加1；
（4）Task：一个Stage阶段中，最后一个RDD的分区个数就是Task的个数。
注意：Application->Job->Stage->Task每一层都是1对n的关系。
4）代码实现
// 创建RDD
val dataRDD: RDD[Int]
= sc.makeRDD(List(1,2,3,4,1,2),2)
//1 聚合
val resultRDD: RDD[(Int, Int)]
= dataRDD.map((_,1)).reduceByKey(_+_)
// Job：一个Action算子就会生成一个Job；
//2 job1打印到控制台
resultRDD.collect().foreach(println)
//3 job2输出到磁盘
resultRDD.saveAsTextFile("output")
Thread.sleep(1000000)
5）查看Job个数
查看http://localhost:4040/jobs/，发现Job有两个。
查看Job0的Stage。由于只有1个Shuffle阶段，所以Stage个数为2。
查看Job1的Stage。由于只有1个Shuffle阶段，所以Stage个数为2。
注意：如果存在shuffle过程，系统会自动进行缓存，UI界面显示skipped的部分
Stage任务划分源码分析
RDD Cache缓存
// 创建一个RDD，读取指定位置文件:hello hadoop hello spark
val lineRdd: RDD[String] = sc.textFile("input1")
val wordRdd: RDD[String] = lineRdd.flatMap(line => line.split(" "))
val wordToOneRdd: RDD[(String, Int)] = wordRdd.map {
println(wordToOneRdd.toDebugString)
// wordToOneRdd.persist(StorageLevel.MEMORY_AND_DISK_2)
println(wordToOneRdd.toDebugString)
def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)
val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
val MEMORY_ONLY = new StorageLevel(false, true, false, true)
val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
注意：默认的存储级别都是仅在内存存储一份。在存储级别的末尾加上"_2"表示持久化的数据存为两份。SER：表示序列化。
// 创建一个RDD，读取指定位置文件:hello hadoop hello spark
val lineRdd: RDD[String] = sc.textFile("input1")
val wordRdd: RDD[String] = lineRdd.flatMap(line => line.split(" "))
val wordToOneRdd: RDD[(String, Int)] = wordRdd.map {
val wordByKeyRDD: RDD[(String, Int)] = wordToOneRdd.reduceByKey(_+_)
println(wordByKeyRDD.toDebugString)
println(wordByKeyRDD.toDebugString)
访问http://localhost:4040/jobs/页面，查看第一个和第二个job的DAG图。说明：增加缓存后血缘依赖关系仍然有，但是，第二个job取的数据是从缓存中取的。
RDD CheckPoint检查点
由于血缘依赖过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果检查点之后有节点出现问题，可以从检查点开始重做血缘，减少了开销。
3）检查点存储路径：Checkpoint的数据通常是存储在HDFS等容错、高可用的文件系统
5）检查点切断血缘：在Checkpoint的过程中，该RDD的所有依赖于父RDD中的信息将全部被移除。
6）检查点触发时间：对RDD进行Checkpoint操作并不会马上被执行，必须执行Action操作才能触发。但是检查点为了数据安全，会从血缘关系的最开始执行一遍。
（1）设置检查点数据存储路径：sc.setCheckpointDir("./checkpoint1")
（2）调用检查点方法：wordToOneRdd.checkpoint()
val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
// 需要设置路径，否则抛异常：__Checkpoint directory has not been set in the SparkContext
val wordRdd: RDD[String] = lineRdd.flatMap(line => line.split(" "))
访问http://localhost:4040/jobs/ 页面，查看4个job的DAG图。其中第2个图是checkpoint的job运行DAG图。第3、4张图说明，检查点切断了血缘依赖关系。
（1）只增加checkpoint，没有增加Cache缓存打印
第1个job执行完，触发了checkpoint，第2个job运行checkpoint，并把数据存储在检查点上。第3、4个job，数据从检查点上直接读取。
第1个job执行完，数据就保存到Cache里面了，第2个job运行checkpoint，直接读取Cache里面的数据，并把数据存储在检查点上。第3、4个job，数据从检查点上直接读取。
缓存和检查点区别
1）Cache缓存只是将数据保存起来，不切断血缘依赖。Checkpoint检查点切断血缘依赖。
2）Cache缓存的数据通常存储在磁盘、内存等地方，可靠性低。Checkpoint的数据通常存储在HDFS等容错、高可用的文件系统，可靠性高。
3）建议对checkpoint()的RDD使用Cache缓存，这样checkpoint的job只需从Cache缓存中读取数据即可，否则需要再从头计算一次RDD。
4）如果使用完了缓存，可以通过unpersist()方法释放缓存
检查点存储到HDFS集群
如果检查点数据存储到HDFS集群，要注意配置访问集群的用户名。否则会报访问权限异常。
val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
val wordRdd: RDD[String] = lineRdd.flatMap(line => line.split(" "))
Spark目前支持Hash分区、Range分区和用户自定义分区。Hash分区为当前的默认分区。分区器直接决定了RDD中分区的个数、RDD中每条数据经过Shuffle后进入哪个分区和Reduce的个数。
（1）只有Key-Value类型的RDD才有分区器，非Key-Value类型的RDD分区的值是None
（2）每个RDD的分区ID范围：0~numPartitions-1，决定这个值是属于那个分区的。
val pairRDD: RDD[(Int, Int)] = sc.makeRDD(List((1,1),(2,2¨C3141C3¨C3142C3¨C3143C
//2 使用HashPartitioner对RDD进行重新分区
val partitionRDD: RDD[(Int, Int)] = pairRDD.partitionBy(new HashPartitioner(2))
println(partitionRDD.partitioner)
Hash分区
HashPartitioner分区的原理：对于给定的key，计算其hashCode，并除以分区的个数取余，如果余数小于0，则用余数+分区的个数（否则加0），最后返回的值就是这个key所属的分区ID。
HashPartitioner分区弊端：可能导致每个分区中数据量的不均匀（数据倾斜），极端情况下会导致某些分区拥有RDD的全部数据。
Ranger分区
第一步：先从整个RDD中采用水塘抽样算法，抽取出样本数据，将样本数据排序，计算出每个分区的最大key值，形成一个Array[KEY]类型的数组变量rangeBounds；
第二步：判断key在rangeBounds中所处的范围，给出该key值在下一个RDD中的分区id下标；该分区器要求RDD中的KEY类型必须是可以排序的
自定义分区
详见章节：RDD编程--Transformation转换算子-- Key-Value类型--自定义分区
数据读取与保存
Spark的数据读取及数据保存可以从两个维度来作区分：文件格式以及文件系统。
文件格式分为：Text文件、Json文件、Csv文件、Sequence文件以及Object文件；
Text文件
（2）数据保存：saveAsTextFile(String)
val inputRDD: RDD[String] = sc.textFile("input/1.txt")
inputRDD.saveAsTextFile("output")
4）注意：如果是集群路径：hdfs://hadoop102:8020/input/1.txt
Sequence文件
val dataRDD: RDD[(Int, Int)] = sc.makeRDD(Array((1,2),(3,4¨C3211C5¨C3212C6¨C3213C
dataRDD.saveAsSequenceFile("output")
sc.sequenceFile[Int,Int]("output").collect().foreach(println)
Object对象文件
val dataRDD: RDD[Int] = sc.makeRDD(Array(1,2,3,4))
dataRDD.saveAsObjectFile("output")
sc.objectFile[(Int)]("output").collect().foreach(println)
累加器
累加器：分布式共享只写变量。（Executor和Executor之间不能读数据）
（1）累加器定义（SparkContext.accumulator(initialValue)方法）
val sum: LongAccumulator = sc.longAccumulator("sum")
val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[1]")
val dataRDD: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("a", 3), ("a", 4)))
注意：Executor端的任务不能读取累加器的值（例如：在Executor端调用sum.value，获取的值不是累加器最终的值）。从这些任务的角度来看，累加器是一个只写变量。
val sum: LongAccumulator = sc.longAccumulator("sum")
val value: RDD[(String, Int)] = dataRDD.map(t => {
//3 调用两次行动算子，map执行两次，导致最终累加器的值翻倍
2）需求：自定义累加器，统计RDD中首字母为"H"的单词以及出现的次数。
List("Hello", "Hello", "Hello", "Hello", "Hello", "Spark", "Spark")
val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
val rdd: RDD[String] = sc.makeRDD(List("Hello", "Hello", "Hello", "Hello", "Spark", "Spark"), 2)
class MyAccumulator extends AccumulatorV2[String, mutable.Map[String, Long]] {
override def copy(): AccumulatorV2[String, mutable.Map[String, Long]] = {
override def merge(other: AccumulatorV2[String, mutable.Map[String, Long]]): Unit = {
广播变量
（1）调用SparkContext.broadcast（广播变量）创建出一个广播对象，任何可序列化的类型都可以这么实现。
（3）变量只会被发到各个节点一次，作为只读值处理（修改这个值不会影响到别的节点）。
采用广播变量的方式，只需要发往每个Executor，每个Executor中的task共用一个副本
val warn: Broadcast[String] = sc.broadcast(list)
val filter: RDD[String] = rdd.filter {
手机扫一扫
移动阅读更方便
你可能感兴趣的文章