Spark常用算子分析与应用

Spark常用算子分析与应用

1、算子概述

什么是算子
- 英文翻译为：Operator（简称op）
- 狭义：指从一个函数空间到另一个函数空间（或它自身）的映射。
- 广义：指从一个空间到另一个空间的映射
- 通俗理解：指事物(数据或函数)从一个状态到另外一个状态的过程抽象。
- 实质就是映射，就是关系，就是变换。
算子的重要作用
- 算子越少，灵活性越低，则实现相同功能的编程复杂度越高，算子越多则反之。
  - 老手机与智能手机
  - 老电脑与新电脑之HDMI (跟VGA是对等的)
- 算子越少，表现力越差，面对复杂场景则易用性较差。算子越多的则反之。
  - 黑白彩电与彩色电视
  - 彩色电视和智能电视
MapReduce 与 Spark算子比较
- MapReduce只有2个算子，Map和Reduce，绝大多数应用场景下，均需要复杂编码才能达到用户需求。
- Spark有80多个算子，进行充分的组合应用后，能满足绝大多数的应用场景。
  2、Spark算子介绍与应用
  2.1 算子分类
  1 转换算子(Transformation）
  此种算子不触发提交作业，只有作业被提交后才会真正启动转换计算。
Value型转换算子：其处理的数据项是Value型
- 输入分区与输出分区一对一型
  - map算子
  - flatMap算子
  - mapPartitions算子
  - glom算子
- 输入分区与输出分区多对一型
  - union算子
  - cartesian算子
- 输入分区与输出分区多对多型
  - groupBy算子
- 输出分区为输入分区子集型
  - filter算子
  - distinct算子
  - subtract算子
  - sample算子
  - takeSample算子
- Cache型
  - cache算子
  - persist算子
Key-Value型转换算子：其处理的数据是Key-Value型
- 输入分区与输出分区一对一
  - mapValues算子
- 对单个RDD聚集
  - combineByKey算子
  - reduceByKey算子
  - partitionBy算子
- 对两个RDD聚集
  - cogroup算子
- 连接
  - join算子
  - leftOutJoin算子
  - rightOutJoin算子
    2 行动算子(Action）
    此种算子会触发SparkContext提交作业。
无输出(是指不输出hdfs、本地文件当中)
- foreach算子
HDFS
- saveAsTextFile算子
- saveAsObjectFile算子
Scala集合和数据类型
- collect算子
- collectAsMap算子
- reduceByKeyLocally算子
- lookup算子
- count算子
- top算子
- reduce算子
- fold算子
- aggregate算子
  2.2 常用算子分析与应用
  2.2.1 Value型转换算子
1) map
- 类比于mapreduce中的map操作，给定一个输入通过map函数映到成一个新的元素输出
  - case_1

val first = sc.parallelize(List(“Hello”,”World”,”天亮教育”,”大数据”),2)
val second= first.map(_.length)
second.collect

  - case_2

val first = sc.parallelize(1 to 5,2)
first.map(1 to _).collect

2) flatMap
- 给定一个输入，将返回的所有结果打平成一个一维集合结构
  - case_1

val first = sc.parallelize(1 to 5,2)
first.flatMap(1 to _).collect

  - case_2

val first = sc.parallelize(List(“one”,”two”,”three”),2)
first.flatMap(x => List(x,x,x)).collect

  - case_3

val first = sc.parallelize(List(“one”,”two”,”three”),2)
first.flatMap(x => List(x+”_1”,x+”_2”,x+”_3”)).collect

3) mapPartitions
- 以分区为单位进行计算处理，而map是以每个元素为单位进行计算处理。
- 当在map过程中需要频繁创建额外对象时，如文件输出流操作、jdbc操作、Socket操作等时，当用mapPartitions算子
  - case_1

val rdd=sc.parallelize(Seq(1,2,3,4,5),3)
var rdd2=rdd.mapPartitions(partition=>{
//在此处加入jdbc等一次初始化多次使用的代码
partition.map(num => num * num)
}
)
rdd2.max

  - case_2

val rdd=sc.parallelize(Seq(1,2,3,4,5),3)
var rdd2=rdd.mapPartitions(partition=>{
//在此处加入jdbc等一次初始化多次使用的代码
partition.flatMap(1 to _)
}
)
rdd2.count

4) glom

以分区为单位，将每个分区的值形成一个数组
val a = sc.parallelize(Seq(“one”,”two”,”three”,”four”,”five”,”six”,”seven”),3)
a.glom.collect

5) union算子
- 将两个RDD合并成一个RDD，并不去重
- 会发生多分区合并成一个分区的情况

val a = sc.parallelize(1 to 4, 2)
val b = sc.parallelize(3 to 6, 2)
a.union(b).collect
(a ++ b).collect
(a union b).collect

6) groupBy算子

输入分区与输出分区多对多型
val a = sc.parallelize(Seq(1,3,4,5,9,100,200), 3)
a.groupBy(x => { if (x > 10) “>10” else “<=10” }).collect

7) filter算子

输出分区为输入分区子集型
val a = sc.parallelize(1 to 21, 3)
val b = a.filter(_ % 4 == 0)
b.collect

8 ) distinct算子

输出分区为输入分区子集型，全局去重

  - case_1

val a = sc.parallelize(1 to 4, 2)
val b = sc.parallelize(3 to 6, 2)
a.union(b).distinct().collect

  - case_2

val c = sc.parallelize(List(“张三”, “李四”, “李四”, “王五”), 2)
c.distinct.collect

9) cache算子
- cache 将 RDD 元素从磁盘缓存到内存。相当于 persist(MEMORY_ONLY) 函数的功能。
- 主要应用在当RDD数据反复被使用的场景下
  - case_1

val a = sc.parallelize(1 to 4, 2)
val b = sc.parallelize(3 to 6, 2)

a.union(b).count
a.union(b).distinct().collect

  - case_2

val a = sc.parallelize(1 to 4, 2)
val b = sc.parallelize(3 to 6, 2)
val c=a.union(b).cache
c.count
c.distinct().collect

2.2.2 Key-Value型转换算子

其处理的数据是Key-Value型

1）mapValues算子
- 输入分区与输出分区一对一
- 针对（Key,Value）型数据中的 Value 进行 Map 操作，而不对 Key 进行处理。

val first = sc.parallelize(List((“张一”,1),(“张二”,2),(“张三”,3),(“张四”,4)),2)
val second= first.mapValues(x=>x+1)
second.collect

2）combineByKey算子
- 定义

def combineByKeyC => C,
mergeValue: (C, V) => C,
mergeCombiners: (C, C) => C): RDD[(String, C)]

  - createCombiner:对每个分区内的同组元素如何聚合，形成一个累加器
  - mergeValue:将前边的累加器与新遇到的值进行合并的方法
  - mergeCombiners:每个分区都是独立处理，故同一个键可以有多个累加器。如果有两个或者更多的分区都有对应同一个键的累加器，用方法将各个分区的结果进行合并。

case_1

val first = sc.parallelize(List((“张一”,1),(“李一”,1),(“张一”,2),(“张一”,3),(“李一”,3),(“李三”,3),(“张四”,4)),2)
val second= first.combineByKey(List(_), (x:List[Int], y:Int) => y :: x, (x:List[Int], y:List[Int]) => x ::: y)
second.collect

3）reduceByKey算子
- 按key聚合后对组进行归约处理，如求和、连接等操作

val first = sc.parallelize(List(“小米”, “华为”, “小米”, “小米”, “华为”, “苹果”), 2)
val second = first.map(x => (x, 1))
second.reduceByKey( + ).collect

4)join算子
- 对Key-Value结构的RDD进行按Key的join操作，最后将V部分做flatmap打平操作。

val first = sc.parallelize(List((“张一”,11),(“李二”,12)),2)
val second = sc.parallelize(List((“张一”,21),(“李二”,22),(“王五”,23)),2)
first.join(second).collect

2.2.3 行动算子(Action）

此种算子会触发SparkContext提交作业。触发了RDD DAG 的执行。
1) 无输出型：不落地到文件或是hdfs的作用
- foreach算子

val first = sc.parallelize(List(“小米”, “华为”, “小米”, “小米”, “华为”, “苹果”), 2)
first.foreach(println _)

2) HDFS输出型
- saveAsTextFile算子

val first = sc.parallelize(List(“小米”, “华为”, “小米”, “小米”, “华为”, “苹果”), 2)
//指定本地保存的目录
first.saveAsTextFile(“file:///home/spark/text“)
//指定hdfs保存的目录，默认路径是hdfs系统中/user/当前用户路径下
first.saveAsTextFile(“spark_shell_output_1”)

Scala集合和数据类型
- 3) collect算子
  - 相当于toArray操作，将分布式RDD返回成为一个scala array数组结果，实际是Driver所在的机器节点，再针对该结果操作

val first = sc.parallelize(List(“小米”, “华为”, “小米”, “小米”, “华为”, “苹果”), 2)
first.collect

4) collectAsMap算子
- 相当于toMap操作，将分布式RDD的kv对形式返回成为一个的scala map集合，实际是Driver所在的机器节点，再针对该结果操作

val first = sc.parallelize(List((“张一”,1),(“李一”,1),(“张一”,2),(“张一”,3),(“李一”,3),(“李三”,3),(“张四”,4)),2)
first.collectAsMap

5)lookup算子
- 对（Key，Value）型的RDD操作，返回指定Key对应的元素形成的Seq。

val first = sc.parallelize(List(“小米”, “华为”, “华米”, “大米”, “苹果”,”米老鼠”), 2)
val second=first.map(x=>({if(x.contains(“米”)) “有米” else “无米”},x))
second.lookup(“有米”)

6) reduce算子
- 先对两个元素进行reduce函数操作，然后将结果和迭代器取出的下一个元素进行reduce函数操作，直到迭代器遍历完所有元素，得到最后结果。

//求value型列表的和
val a = sc.parallelize(1 to 10, 2)
a.reduce( + )
//求key-value型列表的value的和
val a = sc.parallelize(List((“one”,1),(“two”,2),(“three”,3),(“four”,4)), 2)
a.reduce((x,y)=>(“sum”,x._2 + y._2))._2

7) fold算子
- fold算子签名： def fold(zeroValue: T)(op: (T, T) => T): T
- 其实就是先对rdd分区的每一个分区进行op函数，在调用op函数过程中将zeroValue参与计算，最后在对所有分区的结果调用op函数，同理此处zeroValue再次参与计算。

//和是41,公式=(1+2+3+4+5+6+10)+10
sc.parallelize(List(1, 2, 3, 4, 5, 6), 1).fold(10)(+)
//和是51,公式=(1+2+3+10)+(4+5+6+10)+10=51
sc.parallelize(List(1, 2, 3, 4, 5, 6), 2).fold(10)(+)
//和是61,公式=(1+2+10)+(3+4+10)+(5+6+10)+10=61
sc.parallelize(List(1, 2, 3, 4, 5, 6), 3).fold(10)(+)

3、经典算子练习

cartesian算子
subtract算子
sample算子
takeSample算子
persist算子
cogroup算子
leftOuterJoin算子
rightOuterJoin算子
saveAsObjectFile算子
count算子
top算子
aggregate算子

3 Spark常用算子分析与应用

Spark常用算子分析与应用

1、算子概述

2、Spark算子介绍与应用

2.1 算子分类

1 转换算子(Transformation）

2 行动算子(Action）

2.2 常用算子分析与应用

2.2.1 Value型转换算子

2.2.2 Key-Value型转换算子

2.2.3 行动算子(Action）

3、经典算子练习