二、spark--sparkcore原理与使用
[TOC]
成都创新互联长期为上千客户提供的网站建设服务,团队从业经验10年,关注不同地域、不同群体,并针对不同对象提供差异化的产品和服务;打造开放共赢平台,与合作伙伴共同营造健康的互联网生态环境。为湘潭企业提供专业的网站建设、网站制作,湘潭网站改版等技术服务。拥有十载丰富建站经验和众多成功案例,为您定制开发。
一、spark中一些基本术语
RDD:弹性分布式数据集,是spark的核心重点
算子:操作RDD的一些函数
application:用户的写的spark程序(DriverProgram + ExecutorProgram)
job:一个action类算子触发的操作
stage:一组任务,会根据依赖关系将job划分成若干个stage
task:同一个stage内部有多个同样操作的task(但处理的数据不同),是集群中最小的执行单元
可能说完这些概念,其实还是不太懂,没关系,这只是先有点印象
二、RDD基本原理和使用
2.1 什么是RDD
RDD,全称:Resilient Distributed Dataset,也就是弹性分布式数据集。是spark中最基本的数据抽象,它代表一个不可变、可分区、里面的元素可并行计算的集合。RDD具有数据流模型的特点:自动容错、位置感知性调度和可伸缩性。可能这还不清晰,我举个例子:
假设我使用sc.textFile(xxxx)从hdfs中读取一个文件的数据,那么文件的数据就相当于一个RDD,但是事实上这个文件的数据在处理时是处于多个不同的worker节点上的,但是在逻辑上,在这个spark集群,这些数据都是属于一个RDD中的。这也就是为什么说RDD是个逻辑概念,它是整个集群的一个抽象对象,分布在集群中。由此看出,RDD是spark实现数据分布式计算处理的关键。例如:
图 2.1 RDD原理
2.2 RDD的属性
关于RDD的属性,源码中有一段注释,如下:
* Internally, each RDD is characterized by five main properties:
* - A list of partitions
1、是一组分区
理解:RDD是由分区组成的,每个分区运行在不同的Worker上,通过这种方式,实现分布式计算,即数据集的基本组成单位。对于RDD来说,每个分片都会被一个计算任务处理,并决定并行计算的粒度。用户可以在创建RDD时指定RDD的分片个数,如果没有指定,那么就会采用默认值。默认值就是程序所分配到的CPU Core的数目。
* - A function for computing each split
2、split理解为分区
在RDD中,有一系列函数,用于处理计算每个分区中的数据。这里把函数叫做算子。Spark中RDD的计算是以分片为单位的,每个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器进行复合,不需要保存每次计算的结果。
算子类型:
transformation action
* - A list of dependencies on other RDDs
3、RDD之间存在依赖关系。窄依赖,宽依赖。
需要用依赖关系来划分Stage,任务是按照Stage来执行的。RDD的每次转换都会生成一个新的RDD,所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时,Spark可以通过这个依赖关系重新计算丢失的分区数据,而不是对RDD的所有分区进行重新计算。
* - Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)
4、可以自动以分区规则来创建RDD
创建RDD时,可以指定分区,也可以自定义分区规则。
当前Spark中实现了两种类型的分片函数,一个是基于哈希的HashPartitioner,另外一个是基于范围的RangePartitioner。只有对于于key-value的RDD,才会有Partitioner,非key-value的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD本身的分片数量,也决定了parent RDD Shuffle输出时的分片数量。
* - Optionally, a list of preferred locations to compute each split on (e.g. block locations for
* an HDFS file)
5、优先选择离文件位置近的节点来执行任务。
移动计算,不移动数据
这点需要解释下:一般来说spark是构建在hdfs之上,从hdfs中读取数据进行处理的。而hdfs是一个分布式存储,比如有A、B、C三个datanode,假设spark要处理的数据刚好存储在C节点上。如果spark此时将任务放在B节点或者A节点上执行,那么就得先从C节点读取数据,然后经过网络传输到A或B节点,然后才能处理,这其实是很耗费性能。而这里spark的意思是优先在离处理数据比较近的节点上执行任务,也就是优先在C节点上执行任务。这就可以节省数据传输所耗费的时间和性能。也就是移动计算而不移动数据。
2.3 创建RDD
创建RDD首先需要创建 SparkContext对象:
//创建spark配置文件对象.设置app名称,master地址,local表示为本地模式。
//如果是提交到集群中,通常不指定。因为可能在多个集群汇上跑,写死不方便
val conf = new SparkConf().setAppName("wordCount").setMaster("local")
//创建spark context对象
val sc = new SparkContext(conf)
通过sc.parallelize() 创建RDD:
sc.parallelize(seq,numPartitions)
seq:为序列对象,如list,array等
numPartitions:分区个数,可以不指定,默认是2
例子:
val rdd1 = sc.parallelize(Array(1,2,3,4,5),3)
rdd1.partitions.length
通过外部数据源创建
val rdd1 = sc.textFile("/usr/local/tmp_files/test_WordCount.txt")
2.4 算子类型
算子有分为transformation和action类型,
transformation:
延迟计算,lazy懒值,不会触发计算。只有遇到action算子才会触发计算。它们只是记住这些应用到基础数据集(例如一个文件)上的转换动作。只有当发生一个要求返回结果给Driver的动作时,这些转换才会真正运行。这种设计让Spark更加有效率地运行
action:
和transformation类似,但是是直接触发计算的,不会等待
2.5 transformation算子
这里为了方便讲解,实现创建一个rdd,都使用spark-shell中进行演示:
scala> val rdd1 = sc.parallelize(List(1,2,3,4,5,8,34,100,79))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at :24
2.5.1 map(func)
map[U](f:T=>U)()
参数是一个函数,并且要求函数参数是单个,返回值也是单个。用函数处理传入的数据,然后返回处理后的数据
例子:
//这里传入一个匿名函数,将rdd1中的每个值*2,并返回处理的新数组
scala> val rdd2 = rdd1.map(_*2)
rdd2: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[1] at map at :26
//这里collect是一个action算子,触发计算并打印结果
scala> rdd2.collect
res0: Array[Int] = Array(2, 4, 6, 8, 10, 16, 68, 200, 158)
2.5.2 filter
filter(f:T=>boolean)
参数是一个判断函数,判断传入的参数,然后返回true还是false,常用来过滤数据。最后将true的数据返回
例子:
//过滤出大于20的数据
scala> rdd2.filter(_>20).collect
res4: Array[Int] = Array(68, 200, 158)
2.5.3 flatMap
flatMap(f:T=>U)
先map后flat,flat就是将多个列表等对象展开合并成一个大列表。并返回处理后的数据。这个函数一般用来处理多个一个列表中还包含多个列表的情况
例子:
scala> val rdd4 = sc.parallelize(Array("a b c","d e f","x y z"))
rdd4: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[18] at parallelize at :24
//处理逻辑是:将array中每个字符串按空格切割,然后生成多个array,接着将多个array展开合并一个新的array
scala> val rdd5 = rdd4.flatMap(_.split(" "))
rdd5: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[19] at flatMap at :26
scala> rdd5.collect
res5: Array[String] = Array(a, b, c, d, e, f, x, y, z)
2.5.4 集合操作
union(otherDataset) 并集
intersection(otherDataset) 交集
distinct([numTasks]))去重
例子:
scala> val rdd6 = sc.parallelize(List(5,6,7,8,9,10))
rdd6: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[20] at parallelize at :24
scala> val rdd7 = sc.parallelize(List(1,2,3,4,5,6))
rdd7: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[21] at parallelize at :24
//并集
scala> val rdd8 = rdd6.union(rdd7)
rdd8: org.apache.spark.rdd.RDD[Int] = UnionRDD[22] at union at :28
scala> rdd8.collect
res6: Array[Int] = Array(5, 6, 7, 8, 9, 10, 1, 2, 3, 4, 5, 6)
//去重
scala> rdd8.distinct.collect
res7: Array[Int] = Array(4, 8, 1, 9, 5, 6, 10, 2, 7, 3)
//交集
scala> val rdd9 = rdd6.intersection(rdd7)
rdd9: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[31] at intersection at :28
scala> rdd9.collect
res8: Array[Int] = Array(6, 5)
2.5.5 分组操作
groupByKey([numTasks]):只是将同一key的进行分组聚合
reduceByKey(f:(V,V)=>V, [numTasks]) 首先是将同一key的KV进行聚合,然后对value进行操作。
scala> val rdd1 = sc.parallelize(List(("Tom",1000),("Jerry",3000),("Mary",2000)))
rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[32] at parallelize at :24
scala> val rdd2 = sc.parallelize(List(("Jerry",1000),("Tom",3000),("Mike",2000)))
rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[33] at parallelize at :24
scala> val rdd3 = rdd1 union rdd2
rdd3: org.apache.spark.rdd.RDD[(String, Int)] = UnionRDD[34] at union at :28
scala> rdd3.collect
res9: Array[(String, Int)] = Array((Tom,1000), (Jerry,3000), (Mary,2000), (Jerry,1000), (Tom,3000), (Mike,2000))
scala> val rdd4 = rdd3.groupByKey
rdd4: org.apache.spark.rdd.RDD[(String, Iterable[Int])] = ShuffledRDD[35] at groupByKey at :30
//分组
scala> rdd4.collect
res10: Array[(String, Iterable[Int])] =
Array(
(Tom,CompactBuffer(1000, 3000)),
(Jerry,CompactBuffer(3000, 1000)),
(Mike,CompactBuffer(2000)),
(Mary,CompactBuffer(2000)))
注意:使用分组函数时,不推荐使用groupByKey,因为性能不好,官方推荐reduceByKey
//分组并聚合
scala> rdd3.reduceByKey(_+_).collect
res11: Array[(String, Int)] = Array((Tom,4000), (Jerry,4000), (Mike,2000), (Mary,2000))
2.5.6 cogroup
这个函数的功能不太好总结,直接看例子吧
scala> val rdd1 = sc.parallelize(List(("Tom",1),("Tom",2),("jerry",1),("Mike",2)))
rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[37] at parallelize at :24
scala> val rdd2 = sc.parallelize(List(("jerry",2),("Tom",1),("Bob",2)))
rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[38] at parallelize at :24
scala> val rdd3 = rdd1.cogroup(rdd2)
rdd3: org.apache.spark.rdd.RDD[(String, (Iterable[Int], Iterable[Int]))] = MapPartitionsRDD[40] at cogroup at :28
scala> rdd3.collect
res12: Array[(String, (Iterable[Int], Iterable[Int]))] =
Array(
(Tom,(CompactBuffer(1, 2),CompactBuffer(1))),
(Mike,(CompactBuffer(2),CompactBuffer())),
(jerry,(CompactBuffer(1),CompactBuffer(2))),
(Bob,(CompactBuffer(),CompactBuffer(2))))
2.5.7 排序
sortByKey(acsending:true/false) 根据KV中的key排序
sortBy(f:T=>U,acsending:true/false) 一般排序,且是对处理后的数据进行排序,可以用来给KV的,按照value进行排序
例子:
scala> val rdd1 = sc.parallelize(List(1,2,3,4,5,8,34,100,79))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at :24
scala> val rdd2 = rdd1.map(_*2)
rdd2: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[1] at map at :26
scala> rdd2.collect
res0: Array[Int] = Array(2, 4, 6, 8, 10, 16, 68, 200, 158)
scala> rdd2.sortBy(x=>x,true)
res1: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[6] at sortBy at :29
scala> rdd2.sortBy(x=>x,true).collect
res2: Array[Int] = Array(2, 4, 6, 8, 10, 16, 68, 158, 200)
scala> rdd2.sortBy(x=>x,false).collect
res3: Array[Int] = Array(200, 158, 68, 16, 10, 8, 6, 4, 2)
另外一个例子:
需求:
我们想按照KV中的value进行排序,但是SortByKey按照key排序的。
做法一:
1、第一步交换,把key value交换,然后调用sortByKey
2、KV再次调换位置
scala> val rdd1 = sc.parallelize(List(("tom",1),("jerry",1),("kitty",2),("bob",1)))
rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[42] at parallelize at :24
scala> val rdd2 = sc.parallelize(List(("jerry",2),("tom",3),("kitty",5),("bob",2)))
rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[43] at parallelize at :24
scala> val rdd3 = rdd1 union(rdd2)
rdd3: org.apache.spark.rdd.RDD[(String, Int)] = UnionRDD[44] at union at :28
scala> val rdd4 = rdd3.reduceByKey(_+_)
rdd4: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[45] at reduceByKey at :30
scala> rdd4.collect
res13: Array[(String, Int)] = Array((bob,3), (tom,4), (jerry,3), (kitty,7))
//调换位置再排序,然后再调回来
scala> val rdd5 = rdd4.map(t => (t._2,t._1)).sortByKey(false).map(t=>(t._2,t._1))
rdd5: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[50] at map at :32
scala> rdd5.collect
res14: Array[(String, Int)] = Array((kitty,7), (tom,4), (bob,3), (jerry,3))
做法二:
直接使用sortBy 这个算子,可以直接按照value排序
2.6 action算子
reduce
类似前面的reduceByKey,但是用于非KV的数据合并,且是action算子
scala> val rdd1 = sc.parallelize(List(1,2,3,4,5))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[41] at parallelize at :24
scala> val rdd2 = rdd1.reduce(_+_)
rdd2: Int = 15
还有一些action算子:
reduce(func) 通过func函数聚集RDD中的所有元素,这个功能必须是可交换且可并联的
collect() 在驱动程序中,以数组的形式返回数据集的所有元素。通常只是用于触发计算
count() 返回RDD的元素个数
first() 返回RDD的第一个元素(类似于take(1))
take(n) 返回一个由数据集的前n个元素组成的数组
takeSample(withReplacement,num, [seed]) 返回一个数组,该数组由从数据集中随机采样的num个元素组成,可以选择是否用随机数替换不足的部分,seed用于指定随机数生成器种子
takeOrdered(n, [ordering]) ,返回一个由数据集的前n个元素组成的数组,并排序
saveAsTextFile(path) 将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统,对于每个元素,Spark将会调用toString方法,将它装换为文件中的文本
saveAsSequenceFile(path) 将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下,可以使HDFS或者其他Hadoop支持的文件系统。
saveAsObjectFile(path)
countByKey() 针对(K,V)类型的RDD,返回一个(K,Int)的map,表示每一个key对应的元素个数。
foreach(func) 在数据集的每一个元素上,运行函数func进行更新。
2.7 RDD的缓存特性
RDD也存在缓存机制,也就是将RDD缓存到内存或者磁盘中,不用重复计算。
这里涉及到几个算子:
cache() 标识该rdd可以被缓存,默认缓存在内存中,底层其实是调用persist()
persist() 标识该rdd可以被缓存,默认缓存在内存中
persist(newLevel : org.apache.spark.storage.StorageLevel) 和上面类似,但是可以指定缓存的位置
可以缓存的位置有:
val NONE = new StorageLevel(false, false, false, false)
val DISK_ONLY = new StorageLevel(true, false, false, false)
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)
val OFF_HEAP = new StorageLevel(true, true, true, false, 1)
基本就分为两类:
纯内存缓存
纯磁盘缓存
磁盘+内存缓存
一般来说,直接是默认的位置,也就是缓存在内存中性能较好,但是会耗费很多内存,这点要注意,如无需要,就不要缓存了。
举例:
读取一个大文件,统计行数
scala> val rdd1 = sc.textFile("hdfs://192.168.109.132:8020/tmp_files/test_Cache.txt")
rdd1: org.apache.spark.rdd.RDD[String] = hdfs://192.168.109.132:8020/tmp_files/test_Cache.txt MapPartitionsRDD[52] at textFile at :24
scala> rdd1.count
res15: Long = 923452
触发计算,统计行数
scala> rdd1.cache
res16: rdd1.type = hdfs://192.168.109.132:8020/tmp_files/test_Cache.txt MapPartitionsRDD[52] at textFile at :24
标识这个RDD可以被缓存,不会触发计算
scala> rdd1.count
res17: Long = 923452
触发计算,并把结果进行缓存
scala> rdd1.count
res18: Long = 923452
直接从缓存中读取数据。
要注意一个点:调用cache方法的时候,只是说标识该rdd在后续触发计算的时候,结果可以被缓存,而不是说当前rdd就被缓存了,这点要分清楚
2.8 RDD的容错机制--checkpoint
spark在计算时,中间涉及到RDD的多个转换过程,如果这时候RDD的某个分区计算故障,导致结果丢失了。最简单的办法自然是从头开始重新计算,但是这样很浪费时间。而checkpoint就是在触发计算的时候,对RDD进行检查点状态保存,如果后面的计算出错了,还可以从检查点开始重新计算。
checkpoint一般是在具有容错能力,高可靠的文件系统上(比如HDFS, S3等)设置一个检查点路径,用于保存检查点数据。出错的时候直接从检查点目录读取。有本地目录和远程目录两种模式。
2.8.1 本地目录
这种模式下,要求运行在本地模式下,不能用在集群模式下,一般用于测试开发
sc.setCheckpointDir(本地路径) 设置本地checkpoint路径
rdd1.checkpoint 设置检查点
rdd1.count 遇到action类算子,触发计算,就会在checkpoint目录生成检查点
2.8.2 远程目录(hdfs为例)
这种模式要求运行在集群模式下,用于生产环境
scala> sc.setCheckpointDir("hdfs://192.168.109.132:8020/sparkckpt0619")
scala> rdd1.checkpoint
scala> rdd1.count
res22: Long = 923452
用法都是类似,只是目录不一样
要注意,使用checkpoint的时候,源码中有一段话如下:
this function must be called before any job has been executed on this RDD. It is strongly recommend that this rdd is
persisted in memory,otherwise saving it on a file will require recomputation.
大致意思就是:
要在开始计算前就调用这个方法,也就是action算子之前。而且最好设置这个rdd缓存在内存中,否则保存检查点的时候,需要重新计算一次。
2.9 RDD中的依赖与stage原理
2.9.1 RDD依赖
这个是RDD运行原理中的一个重点概念。
首先要说说依赖,依赖的意思就是RDD之间是有依赖关系的,因为spark计算过程中涉及到多个RDD的转换。RDD和它依赖的父RDD(s)的关系有两种不同的类型,即窄依赖(narrow dependency)和宽依赖(wide dependency)。看图
图2.2 RDD的宽窄依赖
宽依赖:
一个父RDD的partition被多个子RDD的partition依赖。其实就是将父RDD的数据进行shuffle的过程,因为父RDD一个partition被多个RDD的partition依赖,意味着需要将父RDD的数据打乱分配给多个RDD,打乱的过程其实就是shuffle。一般实际情况是多个父RDD的partition和多个子RDD的partition互相交错的依赖。
窄依赖:
一个父RDD的partition最多被一个子RDD的partition依赖
2.9.2 stage划分
图2.3 RDD依赖
DAG(Directed Acyclic Graph)叫做有向无环图,原始的RDD通过一系列的转换就就形成了DAG,根据RDD之间的依赖关系的不同将DAG划分成不同的Stage。宽窄依赖的作用就是用划分stage,stage之间是宽依赖,stage内部是窄依赖。
对于窄依赖,由于父和子RDD的partition都是一对一的依赖关系,所以可以父和子的转换可以放在一个task中执行,例如上面的task0,CDF都是窄依赖,所以CDF直接的转换是可以放在一个task中执行的。一个stage内部都是窄依赖
对于宽依赖,由于有shuffle的存在,那么就要求所有父RDD都处理完成后,才能执行执行shuffle,接着子RDD才能进行处理。由于shuffle的存在,导致task任务链必定不是连续的了,需要重新开始规划task任务链,所以宽依赖是划分stage的依据。
再往深的说,为什么要划分stage?
根据宽依赖划分了stage之后,因为宽依赖的shuffle,所以导致task链是无法连续。而划分stage就是让一个stage内部只有窄依赖,窄依赖是一对一的关系,那么task链就是连续的,没有shuffle,就比如上面task0中,C->D->F,中的一个分区,转换过程都是一对一的,所以是一个连续的task链,放在一个task中,而另外一个分区类似,就放在task1中。因为F->G是宽依赖,需要shuffle,所以task链无法连续。像这种一条线把RDD转换逻辑串起来,直到遇到宽依赖,就是一个task,而一个task处理的实际上是一个分区的数据转换过程。而在spark中,task是最小的调度单位,spark会将task分配到离分区数据近的worker节点上执行。所以其实spark调度是task。
那么回到最开始的问题上,为什么要划分stage,因为根据宽窄依赖划分出stage之后,stage内部就可以很方便划分出一个个task,每个task处理一个分区的数据,然后spark就将task调度到对应的worker节点上执行。所以从划分stage到划分task,核心就在于实现并行计算。
所以,最后就是一句话,划分stage的目的是为了更方便的划分task
2.9.3 RDD存储的是什么?
说到这里,其实我们想到一个问题,RDD里面存储的是数据吗?实际上并不是,它存储的实际上对数据的转换链,说的具体点是对分区的转换链,也就是task中包含的算子。而当划分stage,接着划分task之后,一个task内部有什么算子就已经很清楚了,接着就是把计算任务发送到worker节点执行。这种计算我们称为 pipeline计算模式,算子就是在管道中的。
由此,其实RDD叫做弹性分布式数据集,并不是说它存储数据,而是存储了操作数据的方法函数,也就是算子。
2.10 RDD高级算子
2.10.1 mapPartitionsWithIndex
def mapPartitionsWithIndex[U](f: (Int, Iterator[T]) ⇒ Iterator[U])
参数说明:
f是一个函数参数,需要自定义。
f 接收两个参数,第一个参数是Int,代表分区号。第二个Iterator[T]代表该分区中的所有元素。
通过这两个参数,可以定义处理分区的函数。
Iterator[U] : 操作完成后,返回的结果。
举例:
将每个分区中的元素,包括分区号,直接打印出来。
//先创建一个rdd,指定分区数为3
scala> val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8),3)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at :24
scala> def fun1(index:Int,iter:Iterator[Int]):Iterator[String]={
| iter.toList.map( x => "[PartId: " + index + " , value = " + x + " ]").iterator
| }
fun1: (index: Int, iter: Iterator[Int])Iterator[String]
scala> rdd1.mapPartitionsWithIndex(fun1).collect
res0: Array[String] = Array(
[PartId: 0 , value = 1 ], [PartId: 0 , value = 2 ],
[PartId: 1 , value = 3 ], [PartId: 1 , value = 4 ], [PartId: 1 , value = 5 ],
[PartId: 2 , value = 6 ], [PartId: 2 , value = 7 ], [PartId: 2 , value = 8 ]
)
2.10.2 aggregate
聚合操作,类似于分组(group by).
但是aggregate是先局部聚合(类似于mr中的combine),然后再全局聚合。性能比直接使用reduce算子要好,因为reduce是直接全局聚合的。
def aggregate[U](zeroValue: U)(seqOp: (U, T) ⇒ U, combOp: (U, U) ⇒ U)
参数说明:
zeroValue:初始值,这个初始值会加入到每一个分区中,最后也会加入到全局操作中
seqOp: (U, T) ⇒ U:局部聚合操作函数
combOp: (U, U) ⇒ U:全局聚合操作函数
=================================================
例子1:
初始值是10
scala> val rdd2 = sc.parallelize(List(1,2,3,4,5),2)
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[6] at parallelize at :27
//打印看看分区情况
scala> rdd2.mapPartitionsWithIndex(fun1).collect
res7: Array[String] = Array([PartId: 0 , value = 1 ], [PartId: 0 , value = 2 ], [PartId: 1 , value = 3 ], [PartId: 1 , value = 4 ], [PartId: 1 , value = 5 ])
//求出每个分区的最大值,最后得出每个分区最大值,然后全局之后将每个最大值相加
scala> rdd2.aggregate(10)(max(_,_),_+_)
res8: Int = 30
为什么这里是10呢?
初始值是10 代表每个分区都多了一个10
局部操作,每个分区最大值都是10
全局操作,也多一个10 即 10(局部最大) + 10(局部最大) + 10(全局操作默认值) = 30
=================================================
例子2:
使用aggregate将所有分区全局数据求和,有两种方式:
1、reduce(_+_)
2、aggregate(0)(_+_,_+_)
2.10.3 aggregateByKey
类似于aggregate操作,区别:操作的 的数据,只操作同一key的中的value。是将同一key的KV先局部分组,然后对value聚合。然后再全局分组,再对value聚合。
aggregateByKey和reduceByKey实现的类似的功能,但是效率比reduceByKey高
例子:
val pairRDD = sc.parallelize(List(("cat",2),("cat",5),("mouse",4),("cat",12),("dog",12),("mouse",2)),2)
def fun1(index:Int,iter:Iterator[(String,Int)]):Iterator[String]={
iter.toList.map( x => "[PartId: " + index + " , value = " + x + " ]").iterator
}
pairRDD.mapPartitionsWithIndex(fun1).collect
scala> val pairRDD = sc.parallelize(List(("cat",2),("cat",5),("mouse",4),("cat",12),("dog",12),("mouse",2)),2)
pairRDD: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[16] at parallelize at :27
scala> def fun1(index:Int,iter:Iterator[(String,Int)]):Iterator[String]={
| iter.toList.map( x => "[PartId: " + index + " , value = " + x + " ]").iterator
| }
fun1: (index: Int, iter: Iterator[(String, Int)])Iterator[String]
scala> pairRDD.mapPartitionsWithIndex(fun1).collect
res31: Array[String] = Array(
[PartId: 0 , value = (cat,2) ], [PartId: 0 , value = (cat,5) ], [PartId: 0 , value = (mouse,4) ],
[PartId: 1 , value = (cat,12) ], [PartId: 1 , value = (dog,12) ], [PartId: 1 , value = (mouse,2)
])
需求:
找到每个分区中,动物最多的动物,进行就和
pairRDD.aggregateByKey(0)(math.max(_,_),_+_).collect
0:(cat,2)和(cat,5) --> (cat,5) (mouse,4)
1:(cat,12) (dog,12) (mouse,2)
求和:(cat,17) (mouse,6) (dog,12)
2.10.4 coalesce 和 repartition
这两者都用于重分区
repartition(numPartition) 指定重分区个数,一定会发生shuffle
coalesce(numPartition,shuffleOrNot) 指定重分区个数,默认不会发生shuffle,可以指定shuffle
要看更多算子的用法,
写的很详细
2.11 分区
spark自带了两个分区类:
HashPartitioner:这个是默认的partitioner,在一些涉及到shuffle的算子会用到。在一些可以指定最小分区数量的算子中,就会涉及到分区。这些分区只能用于KV对
RangePartitioner:按照key的范围进行分区,比如1~100,101~200分别是不同分区的
用户也可以自己自定义分区,步骤如下:
1、先继承Partitioner类,里面写分区逻辑,形成一个新的分区类
2、rdd.partitionBy(new partiotionerClassxxx())
例子:
数据格式如下:
192.168.88.1 - - [30/Jul/2017:12:55:02 +0800] "GET /MyDemoWeb/web.jsp HTTP/1.1" 200 239
192.168.88.1 - - [30/Jul/2017:12:55:02 +0800] "GET /MyDemoWeb/hadoop.jsp HTTP/1.1" 200 242
需求:
将同一个页面的访问日志各自写到一个单独的文件中
代码:
package SparkExer
import org.apache.spark.{Partitioner, SparkConf, SparkContext}
import scala.collection.mutable
/**
* 自定义分区:
* 1、先继承Partitioner类,里面写分区逻辑,形成一个新的分区类
* 2、rdd.partitionBy(new partiotionerClassxxx())
*/
object CustomPart {
def main(args: Array[String]): Unit = {
//指定hadoop的家目录,写入文件到本地时需要用到hadoop的一些包
System.setProperty("hadoop.home.dir","F:\\hadoop-2.7.2")
val conf = new SparkConf().setAppName("Tomcat Log Partitioner").setMaster("local")
val sc = new SparkContext(conf)
//切割文件
val rdd1 = sc.textFile("G:\\test\\tomcat_localhost_access_log.2017-07-30.txt").map(
line => {
val jspName = line.split(" ")(6)
(jspName,line)
}
)
//提取出所有key,也就是网页名
val rdd2 = rdd1.map(_._1).distinct().collect()
//分区
val rdd3 = rdd1.partitionBy(new TomcatWebPartitioner(rdd2))
//将分区数据写到文件中
rdd3.saveAsTextFile("G:\\test\\tomcat_localhost")
}
}
class TomcatWebPartitioner(jspList:Array[String]) extends Partitioner{
private val listMap = new mutable.HashMap[String,Int]()
var partitionNum = 0
//根据网页名称,规划整个分区个数
for (s<-jspList) {
listMap.put(s, partitionNum)
partitionNum += 1
}
//返回分区总个数
override def numPartitions: Int = listMap.size
//按照key返回某个分区号
override def getPartition(key: Any): Int = listMap.getOrElse(key.toString, 0)
}
2.12 序列化问题
首先我们知道一点,一个spark程序其实是分为两部分的,一部分driver,它也是在executor中运行的,另一部分则是普通的executor,用于运行操作RDD的task的。所以其实也可以看出,只有是对RDD操作的代码才会进行分布式运行,分配到多个executor中运行,但是不属于RDD的代码是不会的,它仅仅是在driver中执行。这就是关键了。
例子:
object test {
val sc = new SparkContext()
print("xxxx1")
val rdd1 = sc.textFile(xxxx)
rdd1.map(print(xxx2))
}
例如上面的例子,rdd1中的print(xxx2)会在多个executor中执行,因为它是在rdd内部执行,而外面的print("xxxx1")只会在driver中执行,也没有实现序列化,所以实际上也不可能通过网络传递。所以这种区别一定要了解。由此我们可以知道,如果变量什么的不是在rdd内部的话,是不可能被多个executor上的程序获得的。但是如果我们想这样呢?而且是不需要定义在rdd内部。那么就得用到下面的共享变量了
2.13 spark中广播变量(共享变量)
广播变量就是可以实现将driver中的变量给在不同的executor中运行的rdd算子调用,而且无需再rdd算子内部定义。常见的比如连接MySQL等数据库的连接对象,可以设置为广播变量,这样就可以只创建一个连接了。
用法例子:
//定义共享变量,用于共享从MongoDB读取的数据,需要将数据封装成 map(mid1,[map(mid2,score),map(mid3,score)....])的形式
val moviesRecsMap = spark.read
.option("uri", mongoConfig.uri)
.option("collection", MOVIES_RECS)
.format("com.mongodb.spark.sql")
.load().as[MoviesRecs].rdd.map(item=> {
(item.mid, item.recs.map(itemRecs=>(itemRecs.mid,itemRecs.socre)).toMap)
}).collectAsMap()
这是关键的一步,就是广播变量出去
//将此变量广播,后面就可以在任意一个executor中调用了
val moviesRecsMapBroadcast = spark.sparkContext.broadcast(moviesRecsMap)
//因为是懒值加载,所以需要手动调用一次才会真正广播出去
moviesRecsMapBroadcast.id
三、spark小案例
3.1 统计访问量前N名的网站页面
需求:根据网站访问日志统计出访问量前N位的网页名称
数据格式如下:
192.168.88.1 - - [30/Jul/2017:12:55:02 +0800] "GET /MyDemoWeb/web.jsp HTTP/1.1" 200 239
192.168.88.1 - - [30/Jul/2017:12:55:02 +0800] "GET /MyDemoWeb/hadoop.jsp HTTP/1.1" 200 242
代码:
package SparkExer
import org.apache.spark.{SparkConf, SparkContext}
/**
* 分析tomcat日志
* 日志例子:
* 192.168.88.1 - - [30/Jul/2017:12:53:43 +0800] "GET /MyDemoWeb/ HTTP/1.1" 200 259
*
* 统计每个页面的访问量
*/
object TomcatLog {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("Tomcat Log analysis").setMaster("local")
val sc = new SparkContext(conf)
val rdd1 = sc.textFile("G:\\test\\tomcat_localhost_access_log.2017-07-30.txt")
.map(_.split(" ")(6))
.map((_,1))
.reduceByKey(_+_)
.map(t=>(t._2,t._1))
.sortByKey(false)
.map(t=>(t._2,t._1))
.collect()
//也可以使用 sortBy(_._2,false)直接根据value进行排序
//取出rdd中的前N个数据
rdd1.take(2).foreach(x=>println(x._1 + ":" + x._2))
println("=========================================")
//取出rdd中的后N个数据
rdd1.takeRight(2).foreach(x=>println(x._1 + ":" + x._2))
sc.stop()
}
}
3.2 自定义分区例子
请看前面2.11中分区的例子
3.3 spark连接mysql
package SparkExer
import java.sql.{Connection, DriverManager, PreparedStatement}
import org.apache.spark.{SparkConf, SparkContext}
object SparkConMysql {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("Tomcat Log To Mysql").setMaster("local")
val sc = new SparkContext(conf)
val rdd1 = sc.textFile("G:\\test\\tomcat_localhost_access_log.2017-07-30.txt")
.map(_.split(" ")(6))
rdd1.foreach(l=>{
//jdbc操作需要包含在rdd中才能被所有worker上的executor调用,也就是借用rdd实现序列化
val jdbcUrl = "jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8"
var conn:Connection = null
//sql语句编辑对象
var ps:PreparedStatement = null
conn = DriverManager.getConnection(jdbcUrl, "root", "wjt86912572")
//?是占位符,后面需要ps1.setxxx(rowkey,value)的形式填充值进去的,按先后顺序
ps = conn.prepareStatement("insert into customer values (?,?)")
ps.setString(1,l)
ps.setInt(2,1)
})
}
}
注意:
spark操作jdbc时,如果直接使用jdbc操作数据库,会有序列化的问题。
因为在spark分布式框架中,所有操作RDD的对象应该是属于RDD内部的,
才有可能在整个分布式集群中使用。也就是需要序列化。
通俗来说:5个worker共享一个jdbc连接对象,和5个worker每个单独创建一个连接对象的区别
所以在定义jdbc连接对象时,需要在RDD内部定义
上面的方式过于繁琐,而且每个数据都会新建一个jdbc连接对象
优化:使用rdd1.foreachPartition()来对每个分区操作,而不是对每条数据操作
这样可以通过只为每个分区创建一个jdbc连接对象来节省数据库资源
package SparkExer
import java.sql.{Connection, DriverManager, PreparedStatement}
import org.apache.spark.{SparkConf, SparkContext}
object SparkConMysql {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("Tomcat Log To Mysql").setMaster("local")
val sc = new SparkContext(conf)
val rdd1 = sc.textFile("G:\\test\\tomcat_localhost_access_log.2017-07-30.txt")
.map(_.split(" ")(6))
rdd1.foreachPartition(updateMysql)
/**
* 上面的方式过于繁琐,而且每个数据都会新建一个jdbc连接对象
* 优化:使用rdd1.foreachPartition()来对每个分区操作,而不是对每条数据操作
* 这样可以通过只为每个分区创建一个jdbc连接对象来节省数据库资源
*/
}
def updateMysql(it:Iterator[String]) = {
val jdbcUrl = "jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8"
var conn:Connection = null
//sql语句编辑对象
var ps:PreparedStatement = null
conn = DriverManager.getConnection(jdbcUrl, "root", "wjt86912572")
//conn.createStatement()
//ps = conn.prepareStatement("select * from customer")
//?是占位符,后面需要ps1.setxxx(rowkey,value)的形式填充值进去的,按先后顺序
ps = conn.prepareStatement("insert into customer values (?,?)")
it.foreach(data=>{
ps.setString(1,data)
ps.setInt(2,1)
ps.executeUpdate()
})
ps.close()
conn.close()
}
}
另外一种连接mysql的方式就是通过JdbcRDD对象去连接
package SparkExer
import java.sql.DriverManager
import org.apache
import org.apache.spark
import org.apache.spark.rdd.JdbcRDD
import org.apache.spark.{SparkConf, SparkContext}
object MysqlJDBCRdd {
def main(args: Array[String]): Unit = {
val conn = () => {
Class.forName("com.mysql.jdbc.Driver").newInstance()
DriverManager.getConnection("jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8",
"root",
"wjt86912572")
}
val conf = new SparkConf().setAppName("Tomcat Log To Mysql").setMaster("local")
val sc = new SparkContext(conf)
//创建jdbcrdd对象
val mysqlRdd = new JdbcRDD(sc,conn,"select * from customer where id>? and id", 2,7,2,r=> {
r.getString(2)
})
}
}
这个对象的使用局限性很大,只能用于select,而且必须传入where中的两个限制值,还要指定分区
四、 shuffle问题
4.1 shuffle导致的数据倾斜分析
https://www.cnblogs.com/diaozhaojian/p/9635829.html
1、数据倾斜原理
(1)在进行shuffle的时候,必须将各个节点上相同的key拉取到某个节点上的一个task来进行处理,此时如果某个key对应的数据量特别大的话,就会发生数据倾斜。
(2)由于shuffle之后的分区规则,导致某个分区数据量过多,导致数据倾斜
2、数据倾斜问题发现与定位
通过Spark Web UI来查看当前运行的stage各个task分配的数据量,从而进一步确定是不是task分配的数据不均匀导致了数据倾斜。
知道数据倾斜发生在哪一个stage之后,接着我们就需要根据stage划分原理,推算出来发生倾斜的那个stage对应代码中的哪一部分,这部分代码中肯定会有一个shuffle类算子。通过countByKey查看各个key的分布。
3、数据倾斜解决方案
过滤少数导致倾斜的key
提高shuffle操作的并行度
局部聚合和全局聚合
4.2 shuffle类算子
1、去重:
def distinct()
def distinct(numPartitions: Int)
2、聚合
def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]
def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)]
def groupBy[K](f: T => K, p: Partitioner):RDD[(K, Iterable[V])]
def groupByKey(partitioner: Partitioner):RDD[(K, Iterable[V])]
def aggregateByKey[U: ClassTag](zeroValue: U, partitioner: Partitioner): RDD[(K, U)]
def aggregateByKey[U: ClassTag](zeroValue: U, numPartitions: Int): RDD[(K, U)]
def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C): RDD[(K, C)]
def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, numPartitions: Int): RDD[(K, C)]
def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, partitioner: Partitioner, mapSideCombine: Boolean = true, serializer: Serializer = null): RDD[(K, C)]
3、排序
def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length): RDD[(K, V)]
def sortBy[K](f: (T) => K, ascending: Boolean = true, numPartitions: Int = this.partitions.length)(implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]
4、重分区
def coalesce(numPartitions: Int, shuffle: Boolean = false, partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null)
5、集合或者表操作
def intersection(other: RDD[T]): RDD[T]
def intersection(other: RDD[T], partitioner: Partitioner)(implicit ord: Ordering[T] = null): RDD[T]
def intersection(other: RDD[T], numPartitions: Int): RDD[T]
def subtract(other: RDD[T], numPartitions: Int): RDD[T]
def subtract(other: RDD[T], p: Partitioner)(implicit ord: Ordering[T] = null): RDD[T]
def subtractByKey[W: ClassTag](other: RDD[(K, W)]): RDD[(K, V)]
def subtractByKey[W: ClassTag](other: RDD[(K, W)], numPartitions: Int): RDD[(K, V)]
def subtractByKey[W: ClassTag](other: RDD[(K, W)], p: Partitioner): RDD[(K, V)]
def join[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, W))]
def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]
def join[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (V, W))]
def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]
文章标题:二、spark--sparkcore原理与使用
文章URL:http://myzitong.com/article/gcodso.html