Spark基础

Spark 基础

Spark 简朴先容

Why Spark

一、MapReduce编程模子的局限性

1、繁杂：只有Map和Reduce两个操作，复杂的逻辑须要大量的样板代码
2、处理效率低：
2.1、Map中间效果写磁盘，Reduce写HDFS，多个Map通过HDFS交换数据
2.2、任务调理与启动开销大
3、不适合迭代处理、交互式处理和流式处理
二、Spark是类Hadoop MapReduce的通用【并行】框架

1、Job中间输出效果可以保存在内存，不再须要读写HDFS
2、比MapReduce平均快10倍以上
三、版本

2014 1.0
2016 2.x
2020 3.x
四、上风

1、速度快
基于内存数据处理，比MR快100个数量级以上（逻辑回归算法测试）
基于硬盘数据处理，比MR快10个数量级以上
2、易用性
支持Java、【Scala】、【Python：pyspark】、R语言
交互式shell方便开发测试
3、通用性
一栈式办理方案：
批处理
交互式查询 spark sql
实时流处理（微批处理） spark streaming
图计算 spark graphic
呆板学习 spark mlib
4、多种运行模式
YARN，生产环境(ApplicationMaster)
Standalone，生产环境(Master/Worker)
Local

，本地模式【初学】√
Spark 技术栈

（图片侵权可删）
1、Spark Core：核心组件，分布式计算引擎 RDD
2、Spark SQL：高性能的基于Hadoop的SQL办理方案
3、Spark Streaming：可以实现高吞吐量、具备容错机制的准实时流处理系统
4、Spark GraphX：分布式图处理框架
5、Spark MLlib：构建在Spark上的分布式呆板学习库
spark-shell：Spark自带的交互式工具

local：spark-shell --master local

alone：spark-shell --master spark://MASTERHOST:7077
yarn ：spark-shell --master yarn
Spark服务（服务上跑的是进程）

Master : Cluster Manager
Worker : Worker Node
Spark架构核心组件

Application ：
建立在Spark上的用户程序，包罗Driver代码和运行在集群各节点Executor中的代码
Driver Program ：
驱动程序，运行Application中的main函数并创建SparkContext
SparkContext ：
Spark应用程序的入口，负责调理各个运算资源，协调各个Worker Node上的Executor
Cluster Manager ：
在集群（Standalone、Mesos、YARN）上获取资源的外部服务
Worker Node ：
集群中任何可以运行Application代码的节点，运行一个或多个Executor进程
Executor ：
Application运行在Worker Node上的一个进程，该进程负责运行Task，并且负责将数据存在内存或者磁盘上
Task ：
运行在Executor上的工作单元
Job ：
多个Task构成的并行计算，由Action触发天生，一个Application中含多个Job
Stage ：
每个Job会被拆分成多组Task，作为一个TaskSet，其名称为Stage

1、在驱动程序中，通过SparkContext主导应用的执行
2、SparkContext可以毗连差别类型的 CM(Standalone、YARN),毗连后,获得节点上的 Executor
3、一个节点默认一个Executor，可通过 SPARK_WORKER_INSTANCES 调解
4、每个应用获取自己的Executor
5、每个Task处理一个RDD分区
SparkContext

1. /*
2. 连接Driver与Spark Cluster（Workers）
3. Spark的主入口
4. 每个JVM仅能有一个活跃的SparkContext
6. 【配置】
7.         master:
8.             local[*] : CPU核数为当前环境的最大值
9.             local[2] : CPU核数为2
10.             local : CPU核数为1
11.             yarn
12. */
13. val conf:SparkConf = new SparkConf()
14.       .setAppName(name:String)
15.           .set(key:String,value:String) // 多项设置
16.       .setMaster(master:String)
17. val sc: SparkContext = SparkContext.getOrCreate(conf)
19. /**
20.         封装：工具类
21. */
22. class SparkCom(appName:String,master:String,logLevel:String="INFO") {
23.   private val conf:SparkConf = new SparkConf().setAppName(appName).setMaster(master)
24.   private var _sc:SparkContext = _
25.   private var _spark:SparkSession = _
26.   def sc() = {
27.     if (Objects.isNull(_sc)) {
28.       _sc = new SparkContext(conf)
29.       _sc.setLogLevel(logLevel)
30.     }
31.     _sc
32.   }
33.   def spark() = {
34.     if (Objects.isNull(_spark)) {
35.       _spark = SparkSession.builder().config(conf).getOrCreate()
36.     }
37.     _spark
38.   }
39.   def stop() = {
40.     if (Objects.nonNull(_sc)) {
41.       _sc.stop()
42.     }
43.     if (Objects.nonNull(_spark)) {
44.       _spark.stop()
45.     }
46.   }
47. }
48. object SparkCom{
49.   def apply(appName:String): SparkCom = new SparkCom(appName,"local[*]")
50.   def apply(appName:String, master:String): SparkCom = new SparkCom(appName,master)
51.   def apply(appName:String, master:String, logLevel:String): SparkCom = new SparkCom(appName,master,logLevel)
52. }

复制代码

RDD

RDD的简朴先容

RDD：Spark核心，重要数据抽象
将数据项拆分为多个分区的集合，存储在集群的工作节点上的内存和【磁盘】
RDD是用于数据转换的接口
RDD指向了
或存储在(HIVE)HDFS、Cassandra、HBase等
或缓存（内存、内存+磁盘、仅磁盘等）
或在故障或缓存收回时重新计算其他RDD分区中的数据
RDD：弹性分布式数据集（Resilient Distributed Datasets）

抽象的：RDD并不存储真正的数据，只是【对数据和操作】的描述
分布式数据集:
RDD是只读的、分区记录的集合，每个分区分布在集群的差别节点上，每个任务处理一个分区，每个分区上都有compute函数，用来计算该分区中的数据
弹性：
RDD默认存放在内存中，当内存不足，Spark自动将RDD写入磁盘
容错性：
根据【数据血统】，可以自动从节点失败中恢复分区
RDD分区：Partition -> Partitioner -> Hash | Range …

分区是RDD被拆分并发送到节点的差别块之一
我们拥有的分区越多，得到的并行性就越强（RDD 是可以并行操作的元素的容错集合）
每个分区都是被分发到差别Worker Node的候选者
每个分区对应一个Task
获取RDD的方式

1. /*
2. 1.小型化数据集
3. 2.读本地文件
4. 3.读远程文件（HDFS）
5. */
6. // 集合创建：小数据集，可通过 numSlices 指定并行度(分区数)
7. val rdd: RDD[T] = sc.parallelize(seq:Seq[T], numSlices:Int) // ✔
8. val rdd: RDD[T] = sc.makeRDD(seq:Seq[T], numSlices:Int) // 调用了 parallelize
10. // 外部数据源创建: 可通过 minPartitions 指定分区数，CPU核占用数
11. // 文件系统：local(file:///...)或hadoop(hdfs://)
12. val rdd: RDD[String] = sc.textFile(path:String, minPartitions:Int)【path】
13. val rdd: RDD[String] = sc.wholeTextFiles(dir:String, minPartitions:Int)【dir】
15. // 其他 RDD 创建
16. val rdd2: RDD[Map[String, Int]] = rdd
17.       .mapPartitions(_
18.         .map(
19.           _
20.             .split("[^a-zA-Z]+")
21.             .map((_, 1))
22.             .groupBy(_._1)
23.             .map(t2 => (t2._1, t2._2.length))
24.         )
25.       )

复制代码

RDD操作类型

分为lazy与non-lazy两种
Transformation（lazy）：也称转换操作、转换算子，返回一个新的RDD
Actions（non-lazy）：立即执行，也称动作操作、动作算子，举措（为）算子，无返回值或者返回其他的
RDD中的全部转换都是惰性求值/延迟执行的，也就是说不会直接进行计算，只有当发生一个要求返回效果给 Driver 的 Action（举措算子） 动作时，这些转换才会真正进行。
之所以使用惰性求值/延迟执行，是因为这样可以在 Action 时对 RDD 操作形成 DAG 有向无环图进行 Stage 的分别和并行优化，这种设计让 Spark 更加有用率的进行。
RDD与DAG： Spark提供的核心抽象

DAG【有向无环图：如下图】反映了RDD之间的依赖关系
转换(transform)算子（窄依赖）==》stage1里面的相互平行
shuffle算子（宽依赖）==》stage2里面的相互交叉
action算子（行为算子）==》末了一个算子为行为算子，不到它不进行计算

（图片侵权可删）
数据预备

1. // 数据类
2. case class Customer(
3.     cus_id            :Int,
4.     cus_fname         :String,
5.     cus_lname         :String,
6.     cus_email         :String,
7.     cus_password      :String,
8.     cus_street        :String,
9.     cus_city          :String,
10.     cus_state         :String,
11.     cus_zipcode       :String
12. )
13. case class OrderItem(
14.     ori_id            :Int,
15.     ori_order_id      :Int,
16.     ori_product_id    :Int,
17.     ori_quantity      :Int,
18.     ori_subtotal      :Float,
19.     ori_product_price :Float
20. )
21. case class Order(
22.     or_id             :Int,
23.     or_date           :String,
24.     or_customer_id    :Int,
25.     or_status         :String
26. )
27. case class Product(
28.     pro_id            :Int,
29.     pro_category_id   :Int,
30.     pro_name          :String,
31.     pro_description   :String,
32.     pro_price         :Float,
33.     pro_image         :String
34. )
36. // 工具
37. implicit class EntityUtil(line:String){
38.     val r_customer = ""(.*?)","(.*?)","(.*?)","(.*?)","(.*?)","(.*?)","(.*?)","(.*?)","(.*?)"".r
39.     val r_product = ""(.*?)","(.*?)","(.*?)","(.*?)","(.*?)","(.*?)"".r
40.     val r_order = ""(.*?)","(.*?)","(.*?)","(.*?)"".r
41.     val r_order_item = ""(.*?)","(.*?)","(.*?)","(.*?)","(.*?)","(.*?)"".r
43.     def toCustomer = line match {
44.         case r_customer(cust_id,fName,lName,email,password,street,city,state,zipcode)
45.                 =>Customer(cust_id.toInt,fName,lName,email,password,street,city,state,zipcode)
46.     }
47.     def toProduct = line match {
48.         case r_product(prod_id,category_id,name,description,price,image)
49.                         =>Product(prod_id.toInt,category_id.toInt,name,description,price.toFloat,image)
50.     }
51.     def toOrder = line match {
52.         case r_order(or_id,date,customer_id,status)
53.                 =>Order(or_id.toInt,date,customer_id.toInt,status)
54.     }
55.     def toOrderItem = line match {
56.         case r_order_item(ori_id,order_id,product_id,quantity,subtotal,product_price)
57.                 =>OrderItem(ori_id.toInt,order_id.toInt,product_id.toInt,quantity.toInt,subtotal.toFloat,product_price.toFloat)
58.     }
59. }

复制代码

转换算子：RDD transform（窄依赖）

1. /*
2.         默认的并行度：200
3.         控制并行度：
4.                 带（并行度） 的算子(执行完计算后对数据进行再分配)
5.                   并行度的体现方式：
6.                         分区数：numParitions|numSlices:Int  5[, fieldName]
7.                                 partitionIndex = fieldName.hashCode() % numParitions
8.                                 扩展随机字段：0~numPartitions
9.                         分区器：partioner:Partioner (key)
10.                                 默认的分区器：HashPartitioner
11.                 再分区算子(对数据进行重新分配)
12.                         coalesce(numPartitions:Int, shuffle:Boolean)
13.                         ✔ repartition(numPartitions:Int) => coalesce(numPartitions, true)
15.         简单类型 RDD[T]
16.         val rddLine: RDD[String] = sc.textFile("/spark/practice_01/customers", 4)
17. */
20. // 【逐条处理】
21. val rdd2: RDD[U] = rdd.map(f:T=>U)
22. val rddAggByState: RDD[(String, Int)] = rddGroupByState.map(t => (t._1, t._2.size))
24. // 【扁平化处理】：TraversableOnce : Trait用于遍历和处理集合类型元素，类似于java:Iterable
25. val rdd2: RDD[U] = rdd.flatMap(f:T=>TraversableOnce[U]) //经典案例为：WordCount
27. /* 【✔ 分区内逐行处理】：以分区为单位(分区不变)逐行处理数据
28.         map VS mapPartitions
29.         1、数量：前者一进一出IO数量一致，后者多进多出IO数量不一定一致
30.         2、性能：前者多分区逐条处理，后者各分区并行逐条处理更佳，常时间占用内存易导致OOM，内存小不推荐
31.         3、类型：两者入口和出口类型都必须一致，后者进出都必须是迭代器
32.        
33.         // 推荐
34.         mapParitions(
35.                 ≈ 子查询
36.                 it.filter(...) <=> 谓词下缀
37.         )
38.        
39.         // 不推荐
40.         mapParitions(...)
41.         fielter(...)   <=> where
42. */
43. val rdd2: RDD[U] = rdd.mapPartitions(f:Iterator[T]=>Iterator[U][,preservePar:Boolean])
44. // 【分区内逐行处理】：以分区为单位(分区不变)逐行处理数据，并追加分区编号
45. val rdd2: RDD[U] = rdd.mapPartitionsWithIndex(f:(Int,Iterator[T])=>Iterator[U][,preservePar:Boolean])
46. val rddCust: RDD[Customer] = rddLine.mapPartitions(_.map(_.toCustomer))
48. // 【转内存数组】：同分区的数据转为同类型的内存数组，分区不变 rdd:RDD[T]
49. val rdd2: RDD[Array[T]] = rdd.glom();
50. val rddCustGlom: RDD[Array[Customer]] = rddCust.glom()
52. // 【× 数据过滤】：过滤规则 f:T=>Boolean，保留满足条件数据，分区不变，【数据可能倾斜skew】
53. val rdd2: RDD[T] = rdd.filter(f:T=>Boolean)
54. val rddFilter: RDD[Customer] = rddCust.filter(_.cus_state.matches("TX|CO|PR"))
56. // 【数据分组】：同键同组同区，同区可多组；打乱shuffle，按f:T=>K规则，分区不变，【数据可能倾斜skew】
57. val rdd2: RDD[(K,Iterable[T])] = rdd.groupBy(f:T=>K)
58. val rdd2: RDD[(K,Iterable[T])] = rdd.groupBy(f:T=>K, partioner:Partitioner)
59. val rdd2: RDD[(K,Iterable[T])] = rdd.groupBy(f:T=>K, numPartitions:Int)
60. val rddGroupByCity: RDD[(String, Iterable[Customer])] = rddFilter.groupBy(_.cus_state)
62. /* 【✔ 数据抽样】
63.         withReplacement:Boolean                是否有放回抽样
64.         fraction:Double                                抽样率
65.         seed:Long                                        随机种子，默认为当前时间戳(一般缺省)
66.       若数据为100条
67.         false, 0.4 => 抽样40%的数据，40条左右
68.         true,  0.4 => 每条数据被抽取的概率为40%
69. */
70. val rdd2: RDD[T] = rdd.sample(withReplacement:Boolean,fraction:Double,seed:Long)
71. val rddSample: RDD[Customer] = rddCust.sample(true, 0.1, 1)
73. // 【× 数据去重】：numPartitions:Int 设定去重后的分区数
74. // 采用该方法去重，数据规模比较大的情况下，数据压力比较大，因为数据需要在不同的分区间比较
75. // 一般采用分组的方式，将去重字段作为分组字段，在不同的分区内并行去重
76. val rdd2: RDD[T] = rdd.distinct()
77. val rdd2: RDD[T] = rdd.distinct(numPartitions:Int)(implicit order:Ording[T] = null)
80. /* 【数据排序】
81.         处理数据f:T=>K，升降序asc:Boolean，分区数numPartitions:Int
82.         默认排序前后分区一致，【有shuffle】，除非重新设定 numPartitions
83.         全局排序，多分区间交换数据，效率较低。优化见 PairRDD
84.         若：K为基本类型，则无需提供第二参数列表中的隐式参数 ord: Ordering[K]
85.         若：K为自定义类型，则必须提供第二参数
86. */
87. val rdd2: RDD[T] = rdd.sortBy(f:T=>K,asc:Boolean,numPartitions:Int)(implicit ord: Ordering[K], ctag: ClassTag[K])
88. val rddSortBy: RDD[Customer] = rddCust.sortBy(c => c.cus_state, asc=true, np=3)
90. /*
91.         多个类型 RDD[T]：纵向
92.                 交并差操作：数据类型一致，根据元素 equals 认定是否相同
93.                         【自定义类型】：必须重写 equals 方法，因为默认等值判断 == 判断地址
94.                 拉链操作：要求分区数和分区内的数据量一致
95. */
96. // 【求交集】：重载可重设分区数numPartitions:Int，或定义分区规则par:Partitioner[T]
97. val rdd2: RDD[T] = rdd.intersection(rdd3:RDD[T])
98. val rdd2: RDD[T] = rdd.intersection(rdd3:RDD[T], numPartitions:Int)
99. val rdd2: RDD[T] = rdd.intersection(rdd3:RDD[T], par:Partitioner[T])
100. // 【求并集】：不去重
101. val rdd2: RDD[T] = rdd.union(rdd3:RDD[T])
102. // 【求差集】：重载可重设分区数numPartitions:Int，或定义分区规则par:Partitioner[T]
103. val rdd2: RDD[T] = rdd.subtract(rdd3:RDD[T])
104. val rdd2: RDD[T] = rdd.subtract(rdd3:RDD[T], numPartitions:Int)
105. val rdd2: RDD[T] = rdd.subtract(rdd3:RDD[T], par:Partitioner[T])
107. // 【拉链操作】
108. val rdd2: RDD[(T,U)] = rdd.zip(rdd3:RDD[U])
109. val rdd2: RDD[(T,Long)] = rdd.zipWithIndex()
110. val rdd2: RDD[(T,Long)] = rdd.zipWithUniqueId()
111. // 有三个重载：1+1，1+2，1+3
112. val rdd2: RDD[V]=rdd.zipPartitions(rddA:RDD[A])(f:(Iterator[T],Iterator[A])=>Iterator[V])
113. val rdd2: RDD[V]=rdd.zipPartitions(rddA:RDD[A],preserveParitioning:Boolean)(f:(Iterator[T],Iterator[A])=>Iterator[V])

复制代码

键值对算子：PairRDD(K,V)

1. // 若再分区器和现有分区器相同，则不执行分区操作
2. val rddPair: RDD[(String, Int)] = rddLine.mapPartitions(it => {
3.     it.map(line => {
4.         val c: Customer = line.toCustomer
5.         (c.cus_state, 1)
6.     })
7. })
9. // 按指定分区器执行再分区操作，根据 K 的自定义 Partitioner 进行，默认为 HashPartitioner
10. val pairRdd2: RDD[(K,V)] = pairRdd.partitionBy(p:Partitioner)
12. // 【按键排序】：
13. /*
14.         排序优化：
15.                 1、自定义分区规则，实现分区内排序
16.                 org.apache.spark.Partitioner
17.             HashPartitioner
18.             RangePargitioner
19.                 2、自定义类型作为 K ，需要提供隐式 Ordering，才可以看到 sortByKey
20.                 3、使用 repartitionAndSortWithinPartitions(par:Partitioner)方法                       
21. */
22. val rddCI: RDD[(Customer, Int)] = rddCust
23.       .mapPartitions(_.map(c=>(c, 1)))
24.           .cache()
26. // 【方法一】
27. val rddCIOrder: RDD[(Customer, Int)] = rddCI
28.       .partitionBy(new CustPartitioner)
29.       .sortBy(_.cust_id)
31. // 【方法二】
32. val pairRdd2: RDD[(K,V)] = pairRdd.sortByKey(ascending:Boolean=true, numPartitions:Int)
34. // 【方法二.一】
35. implicit val orderCust:Ordering[Customer] = Ordering.by(_.cus_id)
36. val rddCIOrder: RDD[(Customer, Int)] = rddCI.sortByKey(true)
38. // 【方法二.二】
39. case class Customer2(
40.     cus_id            :Int,
41.     cus_fname         :String,
42.     cus_lname         :String,
43.     cus_email         :String,
44.     cus_password      :String,
45.     cus_street        :String,
46.     cus_city          :String,
47.     cus_state         :String,
48.     cus_zipcode       :String
49. )
50. extends Ordered[Customer2]
51. with Serializable {
52.     override def compare(that: Customer2): Int =
53.     if(cus_state.compareTo(that.cus_state)==0) cus_id-that.cus_id
54.     else cus_state.compareTo(that.cus_state)
55. }
57. sc.textFile("D:\\projects\\0.2_phrase_hive&spark\\spark_first\\data\\retail_db\\customers.csv", 4)
58.         .mapPartitions(_.map(line=>(line.toCustomer2,1)))
59.         .sortByKey(false)
60.         .foreach(println)
62. // 【方法三】：重新分区，并在当前分区内排序
63. val pairRdd2: RDD[(K,V)] = pairRdd.repartitionAndSortWithinPartitions(par:Partitioner)
65. class CustPartitioner extends Partitioner {
66.     // 分区数：Executor 数量
67.     override def numPartitions: Int = 3
69.     // 根据键获取分区号
70.     override def getPartition(key: Any): Int = key match {
71.         case Customer(_,_,_,_,_,_,_,state,_) => state match {
72.             case "Canada" => 0
73.             case "China" => 1
74.             case "United States" => 2
75.         }
76.     }
77. }
78. implicit val orderCust:Ordering[Customer] = Ordering.by(_.cus_id)
79. val rddCIOrder: RDD[(Customer, Int)] = rddCI
80.       .repartitionAndSortWithinPartitions(new CustPartitioner)
82. // reduceByKey + foldByKey + aggregateByKey 都调用 combineByKeyClassTag
84. // 【✔ 按键聚合值】: combiner和reduce的值类型相同，计算规则相同
85. // group by + combiner + reduce
86. val pairRdd2:RDD[(K,V)] = pairRdd.reduceByKey(f:(V,V)=>V)
87. val pairRdd2:RDD[(K,V)] = pairRdd.reduceByKey(f:(V,V)=>V, numPartitions:Int)
88. val pairRdd2:RDD[(K,V)] = pairRdd.reduceByKey(partitioner:Partitioner, f:(V,V)=>V)
89. val rddPair2: RDD[(String, Int)] = rddPair.reduceByKey(_+_)
91. // 【按键聚合值】: combiner和reduce的值类型相同，计算规则相同，带初值
92. val pairRdd2:RDD[(K,V)] = pairRdd.foldByKey(initV:V)(inParOp:(V,V)=>V)
93. val pairRdd2:RDD[(K,V)] = pairRdd.foldByKey(initV:V,numPartitions:Int)(inParOp:(V,V)=>V)
94. val pairRdd2:RDD[(K,V)] = pairRdd.foldByKey(initV:V,partitioner:Partitioner)(inParOp:(V,V)=>V)
96. // 【✔ 按键分别执行分区内和分区间计算】: combiner和reduce的值类型可不同，计算规则可不同
97. val pairRdd2:RDD[(K,U)] = pairRdd.aggregateByKey(initV:U)(inParOp:(U,V)=>U,betParOp:(U,U)=>U)
98. val pairRdd2:RDD[(K,U)] = pairRdd.aggregateByKey(initV:U,numPartitions:Int)(inParOp:(U,V)=>U,betParOp:(U,U)=>U)
99. val pairRdd2:RDD[(K,U)] = pairRdd.aggregateByKey(initV:U,partitioner:Partitioner)(inParOp:(U,V)=>U,betParOp:(U,U)=>U)
100. val rddPair2: RDD[(String, Float)] = rddPair.aggregateByKey(0.0f)(_+_,_+_)
102. // 【按键分别执行分区内和分区间计算】: combiner和reduce的值类型可不同，计算规则可不同
103. val pairRdd2:RDD[(K,U)] = pairRdd.combineByKey(initV:V=>U,inParOp:(U,V)=>U,betParOp:(U,U)=>U)
104. val pairRdd2:RDD[(K,U)] = pairRdd.combineByKey(initV:V=>U,inParOp:(U,V)=>U,betParOp:(U,U)=>U,numPartitions:Int)
105. val pairRdd2:RDD[(K,U)] = pairRdd.combineByKey(initV:V=>U,inParOp:(U,V)=>U,betParOp:(U,U)=>U,partitioner:Partitioner,mapSideCombine:Boolean,serializer:Serializer)
107. // 【✔ 按键分组】
108. val pairRdd2: RDD[(K, Iterable[V])] = pairRdd.groupByKey()
109. val pairRdd2: RDD[(K, Iterable[V])] = pairRdd.groupByKey(numPartitions:Int)
110. val pairRdd2: RDD[(K, Iterable[V])] = pairRdd.groupByKey(partitioner:Partitioner)
112. // 【多数据集分组】：1VN 同键同组，不同RDD值进入TupleN的不同Iterable
113. -------------------------------------------------------------------------------
114. val pairRdd2: RDD[(K, (Iterable[V],Iterable[V1])] = pairRdd.groupWith(otherA: RDD[(K,V1)])
115. val pairRdd2: RDD[(K, (Iterable[V],Iterable[V1],Iterable[V2])] = pairRdd.groupWith(otherA: RDD[(K,V1)],otherB: RDD[(K,V2)])
116. val pairRdd2: RDD[(K, (Iterable[V],Iterable[V1],Iterable[V2],Iterable[V3])] = pairRdd.groupWith(otherA: RDD[(K,V1)],otherB: RDD[(K,V2)],otherC: RDD[(K,V3)])
117. -------------------------------------------------------------------------------
118. // 重载 1+1 1+2 1+3，追加再分区操作
119. val pairRdd2: RDD[(K, (Iterable[V],Iterable[V1])] = pairRdd.cogroup(otherA: RDD[(K,V1)])
120. val pairRdd2: RDD[(K, (Iterable[V],Iterable[V1])] = pairRdd.cogroup(otherA: RDD[(K,V1)],numPartitions:Int)
121. val pairRdd2: RDD[(K, (Iterable[V],Iterable[V1])] = pairRdd.cogroup(otherA: RDD[(K,V1)],partitioner:Partitioner)
123. // 每国家有多少人？
124. val rddAvgByState: RDD[(String, Float)] = rddLine.mapPartitions(it => {
125.     it.map(line => {
126.         val c: Customer = line.toCustomer
127.         (c.cus_state, 1)
128.     })
129. })
130. .reduceByKey(_ + _)
132. // 每个国家平均每个城市有多少人？
133. val rddAvgByState: RDD[(String, Float)] = rddLine.mapPartitions(it => {
134.     it.map(line => {
135.         val c: Customer = line.toCustomer
136.         (s"${c.cus_state}_${c.cus_city}", 1)
137.     })
138. })
139. .reduceByKey(_ + _)
140. .mapPartitions(it => {
141.     it.map(t => {
142.         val ps: Array[String] = t._1.split("_")
143.         (ps(0), t._2)
144.     })
145. })
146. .groupByKey()
147. .mapPartitions(_.map(t=>(t._1,t._2.sum.toFloat/t._2.size)))
149. /*
150.         【关联操作】：1V1                        Shuffle ?
151.                 横向，根据键做关联
152.                 重载：numPartitions:Int 或 partitioner:Partitioner
153. */
154. val pairRdd: RDD[(K, (V, V1))] = pairRdd1.join(pairRdd3:RDD[(K,V1)])
155. val pairRdd: RDD[(K, (V, Option[V1]))] = pairRdd1.leftOuterJoin(pairRdd3:RDD[(K,V1)])
156. val pairRdd: RDD[(K, (Option[V]), V1)] = pairRdd1.rightOuterJoin(pairRdd3:RDD[(K,V1)])
157. val pairRdd: RDD[(K, (Option[V]), Option[V1])] = pairRdd1.fullOuterJoin( pairRdd3:RDD[(K,V1)])
159.     val rddCus: RDD[Customer] =
160.       sc.textFile("D:\\projects\\0.2_phrase_hive&spark\\spark_first\\data\\retail_db\\customers.csv", 4)
161.         .mapPartitions(_.map(_.toCustomer)).cache()
163.     val rddOrder: RDD[(Int,Order)] =
164.       sc.textFile("D:\\projects\\0.2_phrase_hive&spark\\spark_first\\data\\retail_db\\orders.csv", 4)
165.         .mapPartitions(_.map(line=>{
166.           val order: Order = line.toOrder
167.           (order.or_customer_id,order)
168.         }))
169.         .cache()
171.     // 用户的订单数量
172.     val rddCusOrd: RDD[(Int, Int)] = rddCus.mapPartitions(
173.       _.map(c => (c.cus_id, c))
174.     ).leftOuterJoin(rddOrder)
175.       .mapPartitions(_.map(t => (t._1, if (t._2._2.isEmpty) 0 else 1)))
176.       .reduceByKey(_ + _)
178.         // 用户订单金额
179.         val rddOI: RDD[(Int,Float)] =
180.       sc.textFile("D:\\projects\\0.2_phrase_hive&spark\\spark_first\\data\\retail_db\\order_items", 4)
181.         .mapPartitions(_.map(line=>{
182.           val oi: OrderItem = line.toOrderItem
183.           (oi.ori_order_id,oi.ori_subtotal)
184.         }))
185.         .cache()
186.        
187.         val rddCusSum: RDD[(Int, Float)] = rddCus.mapPartitions(
188.       _.map(c => (c.cus_id, c))
189.     ).leftOuterJoin(rddOrder)
190.       .mapPartitions(_.map(t => (if (t._2._2.isEmpty) 0 else t._2._2.get.or_id, t._1)))
191.       .leftOuterJoin(rddOI)
192.       .mapPartitions(_.map(t => (t._2._1, if (t._2._2.isEmpty) 0 else t._2._2.get)))
193.       .reduceByKey(_ + _)

复制代码

举措算子：action

1. /* 【返回】所有元素分别在分区间和分区内执行【聚集】操作的结果
2.         reduce & fold 分区内和分区间执行相同操作，且类型与元素类型一致
3.         aggregate 分区内和分区间执行不同操作，且类型与元素类型不一致
4. */
5. val rst:T = rdd.reduce(f:(T,T)=>T)
6. val rst:T = rdd.fold(init:T)(f:(T,T)=>T)
7. val rst:U = rdd.aggregate(init:U)(f:(U,T)=>U,f:(U,T)=>U)
8. // 返回包含数据集中所有元素的数组
9. val array:Array[T] = rdd.collect()
10. // 返回数据集中元素数量
11. val rst:Long = rdd.count()
12. val rst:Map[K,Long] = pairRdd.countByKey()
13. // 返回数据集中最大值
14. val rst:T = rdd.max()
15. // 返回数据集中最小值
16. val rst:T = rdd.min()
17. // 返回数据集中的第一个元素
18. val rst:T = rdd.first()
19. // 返回数据集中的前 num 个元素
20. val array:Array[T] = rdd.take(num:Int)
21. // 返回排序后数据集中的前 num 个元素
22. val array:Array[T] = rdd.takeOrdered(num:Int)(implicit ord:Ordering[T])
23. /* 持久化至文本文件，重载追加压缩功能
24.         import org.apache.hadoop.io.compress.{BZip2Codec, SnappyCodec}
25.         import io.airlift.compress.lzo.LzopCodec
26.         rdd.saveAsTextFile("out_path",classOf[BZip2Codec])
27. */
28. rdd.saveAsTextFile(path:String)
29. rdd.saveAsTextFile(path:String,codec: Class[_ <: CompressionCodec])
30. rdd.saveAsObjectFile(path:String)
31. // 遍历迭代
32. rdd.foreach(f:T=>Unit)

复制代码

经典案例（wordCount《==》单词计数功能）

1. // 工具类的创建（SparkUtil，实现一劳永逸）
2. package spark_util
4. import org.apache.spark.sql.SparkSession
5. import org.apache.spark.{SparkConf, SparkContext}
7. class SparkUtil {
8.   /**
9.    *  spark configuration:配置工具类
10.    *  appName,master,logLevel,sparkHome
11.    */
12.   private var config:SparkConf = null
13.   /**
14.    *  spark rdd:弹性分布式数据集（rdd算子）
15.    */
16.   private var sc:SparkContext = null
17.   /**
18.    *  spark sql:结构化查询（sql算子，sql语句）
19.    */
20.   private var spark:SparkSession = null
22.   private def check(info:String,value:Any,regex:String=".*")={
23.     if(null==value){
24.       throw new RuntimeException(s"$info 空指针异常")
25.     }
26.     if(value.isInstanceOf[String] && value.toString.size==0){
27.       throw new RuntimeException(s"$info 空字符串异常")
28.     }
29.     if (null!=regex && !value.toString.matches(regex)) {
30.       throw new RuntimeException(s"$info 不符合正则规则 $regex 异常: ${value.toString}")
31.     }
32.   }
34.   def build()={
35.     config = new SparkConf()
36.     this
37.   }
38.   def appName(name:String) ={
39.     check("SparkConfig",config)
40.     check("appName",name,"[a-z]\\w+")
41.     config.setAppName(name)
42.     this
43.   }
45.   /**
46.    * @param master
47.    * 1、local:单线程
48.    * 2、local[*]:所有可用线程
49.    * 3、local[N]:N条线程
50.    * 4、spark://host:7077: standalone
51.    * 5、yarn-client:yarn模式，本地化运行
52.    * 6、yarn-cluster:yarn模式，ApplicationMaster
53.    */
54.   def master(master:String)={
55.     check("SparkConfig",config)
56.     check("appName",master,
57.       "local\\[(\\*|\\d+)]|spark://[a-z][a-z0-9]*:\\d+|yarn-(client|cluster)")
58.     config.setMaster(master)
59.     this
60.   }
62.   def set(item:String,value:String)={
63.     check("SparkConfig",config)
64.     check("ConfigItemName",item)
65.     check(item,value)
66.     config.set(item,value)
67.   }
69.   def sparkContext(level:String="INFO")={
70.     if(null==sc){
71.       check("SparkConfig",config)
72.       check("LogLevel",level.toUpperCase, "INFO|DEBUG|WARN|ERROR|FATAL")
73.       sc = new SparkContext(config)
74.       sc.setLogLevel(level)
75.     }
76.     sc
77.   }
79.   def sparkSession(hiveSupport:Boolean=false,level:String="INFO")={
80.     if(null==spark){
81.       check("SparkConfig",config)
82.       check("LogLevel",level.toUpperCase, "INFO|DEBUG|WARN|ERROR|FATAL")
83.       // 在创建 SparkSession 之前先创建 SparkContext
84.       spark ={
85.         if(hiveSupport)
86.           SparkSession
87.           .builder()
88.           .config(config)
89.           .enableHiveSupport()
90.           .getOrCreate()
91.         else
92.           SparkSession
93.           .builder()
94.           .config(config)
95.           .getOrCreate()
96.       }
97.       sc = spark.sparkContext
98.     }
99.     spark
100.   }
102.   def close={
103.     if (null != spark) {
104.       // SparkSession 和 SparkContext 并存
105.       // 关闭 SparkSession 同时自动关闭 SparkContext
106.       spark.stop()
107.     }else if(null != sc){
108.       sc.stop()
109.     }
110.   }
111. }
113. object SparkUtil{
114.   def apply(): SparkUtil = new SparkUtil()
115. }
117. // wordCount 具体实现代码（ SparkRDD ）
118. // 读取一个文本文件，统计每个单词出现的次数，并将结果按单词出现次数降序排序
119. package spark_rdd
121. import org.apache.spark.{Partitioner, SparkContext}
122. import spark_util.SparkUtil
123. import spark_util.Data._
125. object SparkRDD {
126.   def main(args: Array[String]): Unit = {
127.            val sparkUtil: SparkUtil = SparkUtil()
128.                 .build()
129.                 .appName("spark_rdd_01")
130.                 .master("local[4]")
131.                
132.     val sc: SparkContext = sparkUtil.sparkContext(level = "ERROR")
133.    
134.         // 从指定路径读取文本文件story.txt，并设置并行度为2。这意味着Spark将尝试使用两个分区来处理文件
135.         sc.textFile("D:\\projects\\0.2_phrase_hive&spark\\spark_first\\data\\story.txt",2)
136.                 // .mapPartitions(it=> { ... })：
137.                 // 一个转换操作，mapPartitions允许你处理每个分区的数据。参数it是一个Iterator[String]，表示每个分区中的行
138.                 .mapPartitions(it=>{
139.                 // it.flatMap(_.trim.replaceAll("[^a-zA-Z ]+","").split("\\s+").map((_,1))：
140.                         /**
141.                                 flatMap：将每个元素转换为多个元素
142.                                 trim：去除每行的前后空白字符
143.                                 replaceAll("[^a-zA-Z ]+","")：删除行中除了字母和空格之外的所有字符
144.                                 split("\\s+")：按空白字符分割字符串，生成单词数组
145.                                 map((_,1))：将每个单词与常量1配对，形成一个二元组
146.                         */
147.                   it.flatMap(         
148.                     _.trim            
149.                                         .replaceAll("[^a-zA-Z ]+","")
150.                                         .split("\\s+")
151.                                         .map((_,1))      
152.                         ).toArray                                // 将 Iterator 转换为数组
153.           // 分区内聚合 : Combiner
154.                         .groupBy(_._1)                        // 按二元组的第一个元素进行分组
155.                         .mapValues(_.size)                // 计算每个单词出现的次数
156.                         .toIterator                                // 奖结果转换回 Iterator
157.                 }).groupBy(_._1,4)                        // 再次按单词进行分组，这次指定了4个分区。这有助于在后续步骤中减少数据的shuffle
158.                 // 再次使用mapPartitions来处理每个分区的数据
159.         .mapPartitions(it=>{
160.                         it.map(tp2=>{
161.                                 val word: String = tp2._1
162.                                 // Reducer : 不同分区同一个单词出现的次数，将不同分区同一个单词的次数求和
163.                                 val count: Int = tp2._2   // Iterator[(String,Int)]
164.                                         .map(_._2)              // Iterator[Int]
165.                                         .sum
166.                                 (word,count)
167.                         }).toArray
168.                                 .sortWith(_._2>_._2)  // (a:(String,Int),b:(String,Int))=>{...}        按单词出现的次数进行降序排序
169.                                 .toIterator
170.         }).sortBy(_._2,false)
171.         .saveAsTextFile("file:///D:\\projects\\0.2_phrase_hive&spark\\spark_first\\data\\result\\wc")
172.   }
173. }

复制代码

RDD专项练习

1. /*
2.         现有客户信息文件 customers.csv，请找出：
3.         客户中的前5个最大家族
4.         客户中的前10个最流行的名字
5. */
7. /*
8.         现有客户信息文件 scores.txt，请找出：
9.         班级 ID 姓名 年龄 性别 科目 成绩
10.         需求如下：
11.     1. 一共有多少人参加考试？
12.         1.1 一共有多少个小于 20 岁的人参加考试？
13.         1.2 一共有多少个等于 20 岁的人参加考试？
14.         1.3 一共有多少个大于 20 岁的人参加考试？
15.     2. 一共有多个男生参加考试？
16.         2.1 一共有多少个女生参加考试？
17.     3. 12 班有多少人参加考试？
18.         3.1 13 班有多少人参加考试？
19.     4. 语文科目的平均成绩是多少？
20.         4.1 数学科目的平均成绩是多少？
21.         4.2 英语科目的平均成绩是多少？
22.     5. 单个人平均成绩是多少？
23.     6. 12 班平均成绩是多少？
24.         6.1 12 班男生平均总成绩是多少？
25.         6.2 12 班女生平均总成绩是多少？
26.         6.3 同理求 13 班相关成绩
27.     7. 全校语文成绩最高分是多少？
28.         7.1 12 班语文成绩最低分是多少？
29.         7.2 13 班数学最高成绩是多少？
30.     8. 总成绩大于 150 分的 12 班的女生有几个？
31. */
33. package spark_rdd2
35. import org.apache.spark.rdd.RDD
36. import org.apache.spark.util.LongAccumulator
37. import org.apache.spark.{SparkConf, SparkContext}
39. object SparkRDD {
40.   // 样例类参与 RDD 运算不能写在 main 中，否则报错：序列化异常
41.   case class Score(classId:Int,name:String,age:Int,gender:String,subject:String,score:Int,_type:String)
42.   def main(args: Array[String]): Unit = {
43.     val conf: SparkConf = new SparkConf()
44.                 .setAppName("spark")
45.                 .setMaster("local[4]")
47.     val sc: SparkContext = SparkContext.getOrCreate(conf)
48.     // 设置检查点路径
49.     sc.setCheckpointDir("hdfs://single01:9000/spark/checkpoint")
51.     val map = Map(
52.                 "GT20"->"20岁以上",
53.                 "EQ20"-> "20岁",
54.                 "LT20"-> "20岁以下"
55.     )
57. //    val mapBroad: Broadcast[Map[String, String]] = sc.broadcast(map)
58.     val maleCount: LongAccumulator = sc.longAccumulator("maleCount")
60.     val path = "file:///D:\\projects\\0.2_phrase_hive&spark\\spark_first\\data\\scores.txt"
61.     val scores: RDD[Score] = sc
62.                 .textFile(path, 3)
63.                 .mapPartitions(
64.                         _.drop(1)
65.                                 .map(line=>{
66.                                         val a: Array[String] = line.split("\\s+")
67.                                         val age: Int = a(2).toInt
68.                                         val _type = age match {
69.                                                 case age if (age > 20) => map.get("GT20").get
70.                                                 case age if (age == 20) => map.get("EQ20").get
71.                                                 case _ => map.get("LT20").get
72.                                         }
73.                                         if(a(3).equals("男")){
74.                                                 maleCount.add(1)
75.                                         }
76.                             Score(a(0).toInt,a(1),age,a(3),a(4),a(5).toInt,_type)
77.                           })
78.                 )
79.       .cache()
81.     scores.foreach(println)
83.     println(maleCount.value)
85.     /**
86.      *   mapPartitions接收一个函数，该函数将每个分区中的迭代器转换为另一个迭代器(出入相同，都为Iterator)
87.      *   groupByKey方法的作用是将具有相同键的所有值聚集在一起
88.      *   reduceByKey方法用于合并具有相同键的所有值
89.      */
91.     // 一共有多少人参加考试（小于20，等于20，大于20）
92.     val personCount: Long = scores
93.                 .mapPartitions(_.map(_.name))
94.                 .distinct()
95.                 .count()
96.     println(s"$personCount 人考试")
97.     scores
98.                 .mapPartitions(_.map(s=>(s._type,1)))
99.                 .reduceByKey(_+_)
100.                 .foreach(println)
102.     // 一共有多少男生（女生）参加考试
103.     val genderCount: RDD[(String, Int)] = scores
104.                 .mapPartitions(_.map(s => (s.gender, 1)))
105.                 .reduceByKey(_ + _)
106.     genderCount.checkpoint()
107.     genderCount.foreach(println)
109.     // 12班（13班）一共有多少人参加考试
110.     scores
111.                 .mapPartitions(_.map(s=>((s.classId,s.name),1)))
112.                 .groupByKey()
113.                 .mapPartitions(_.map(t=>(t._1._1,1)))
114.                 .reduceByKey(_+_)
115.                 .foreach(println)
117.     scores
118.                 .mapPartitions(_.map(t=>(t.classId,t.name)).toSet[(Int,String)].toIterator.map(t=>(t._1,1)))
119.                 .reduceByKey(_+_)
120.                 .foreach(println)
122.     // 语数外各科目的平均成绩
123.     scores
124.                 .mapPartitions(_.map({
125.                 case s if s.subject matches("chinese|math|english") => (s.subject,s.score)
126.                 }))
127.                 .groupByKey()
128.                 .mapPartitions(_.map(t=>(t._1,t._2.sum*1.0f/t._2.size)))
129.                 .reduceByKey(_+_)
130.                 .foreach(println)
132.     // 单个人的平均成绩
133.     scores
134.                 .mapPartitions(_.map(t=>(t.name,t.score)))
135.                 .groupByKey()
136.                 .mapPartitions(_.map(t=>(t._1,t._2.sum*1.0f/t._1.size)))
137.                 .reduceByKey(_+_)
138.                 .foreach(println)
140.     // 12班（13）平均总成绩
141.     scores
142.                 .mapPartitions(_.map(t=>(t.classId,t.score)))
143.                 .groupByKey()
144.                 .mapPartitions(_.map(t=>(t._1,t._2.sum*1.0f/t._2.size)))
145.                 .reduceByKey(_+_)
146.                 .foreach(println)
148.     // 12班（13班）男生（女生）平均总成绩
149.     scores
150.                 .mapPartitions(_.map(t=>((t.classId,t.gender),t.score)))
151.                 .groupByKey()
152.                 .mapPartitions(_.map(t=>(t._1,t._2.sum*1.0f/t._2.size)))
153.                 .reduceByKey(_+_)
154.                 .foreach(println)
156.     // 全校语文成绩最高分
157.     val maxScore: Int = scores
158.                 .mapPartitions(_.filter(_.subject.equals("chinese")).map(_.score))
159.                 .max()
160.     println(s"全校语文成绩最高分为 $maxScore")
162.     // 12 班语文成绩最高分
163.     val chineseScore: RDD[(Int, Int)] = scores
164.                 .mapPartitions(_.filter(_.subject == "chinese").map(t => (t.classId, t.score)))
165.     val chineseMaxScore: Int = chineseScore.mapPartitions(_.filter(_._1 == 12).map(_._2)).max()
166.     println(s"12 班语文成绩最高分为 $chineseMaxScore")
168.     // 13 班数学成绩最高分
169.     val mathScore: RDD[(Int, Int)] = scores
170.                 .mapPartitions(_.filter(_.subject == "math").map(t => (t.classId, t.score)))
171.     val mathMaxScore: Int = mathScore.mapPartitions(_.filter(_._1 == 13).map(_._2)).max()
172.     println(s"13 班数学成绩最高分为 $mathMaxScore")
174.     // 总成绩大于 150 分 的 12 班的女生有几个
175.     // 方法一
176.     val BigScores: RDD[((Int, String), Iterable[Int])] = scores.mapPartitions(_.map(t => ((t.classId, t.gender), t.score)))
177.                 .groupByKey()
178.                 .mapPartitions(_.map({
179.                         case t if t._2.sum > 150 => (t._1, t._2)
180.                 }))
181.     val BigGenderScore: RDD[(Int, Iterable[Int])] = BigScores.mapPartitions(_.filter(_._1._2 == "女")).map(t => (t._1._1, t._2))
183.     val resultScore: Long = BigGenderScore.mapPartitions(_.filter(_._1 == 12)).map(_._2).count()
184.     println(s"总成绩大于 150 分的 12 班的女生有 $resultScore 个")
186.         val numSumScore12Gt150: Long = scores
187.                 .filter(score => score.classId == 12 && score.gender.equals("女"))
188.                 .mapPartitions(_.map(score => (score.name, score.score)))
189.                 .reduceByKey(_+_)
190.                 .filter(_._2 > 150)
191.                 .count()
193.     sc.stop()
194.   }
195. }

复制代码

优化：optimize

1. org.apache.spark.util.Utils
2.        
3. /*
4.         shuffle性能较差：因为shuffle必须落盘，内存中等数据会OOM
5.         groupByKey只分组(存在Shuffle) + reduce只聚合
6.                 <=结果同，性能不同=>
7.         reduceByKey先分组、预聚合、再聚合(存在Shuffle) ✔
8. */
10. /*
11. 【设置日志管理】
12.         日志级别：INFO|DEGUG|WARN|ERROR|FATAL
13. */
14. sc.setLogLevel(logLevel:String)
16. /*
17. 【设置检查点：容错，恢复】
18. */
19. sc.setCheckpointDir(path:String)
21. /*
22. 【RDD重用：检查点、缓存与持久化】
23.         cache      临时存储于【内存】重用，job结束后自动删除 ✔
24.                 cache  相当于 persist(StorageLevel.MEMORY_ONLY)
25.         persisit   临时存储于【磁盘】重用，job结束后自动删除，涉及IO性能较差
26.                 StorageLevel.MEMORY_ONLY
27.         StorageLevel.DISK_ONLY
28.         StorageLevel.OFF_HEAP
29.         StorageLevel.MEMORY_AND_DISK
30.         StorageLevel.MEMORY_AND_DISK_SER
31.         StorageLevel.MEMORY_AND_DISK_SER_2
32.         checkpoint 长久存储于【磁盘】重用，job结束后不会删除，涉及IO性能较差，安全且一般和cache组合使用
33. */
34. val rddCache: RDD[T] = rdd.cache()
35. val rddCache: RDD[T] = rdd.persist(level:StorageLevel)
36. rdd.checkpoint()
38. /*
39. 【广播变量】（分布式只读变量）：broadcast
40. 作用：
41. 广播变量用来把变量在所有节点的内存之间进行共享，在每个机器上缓存一个只读的变量，而不是为机器上的每个任务都生成一个副本
43. 将数据集或配置广播到每个Executor以readonly方式存在，不会在Task之间传输
45. 优势：
46. 如果Executor端需要访问Driver端的某个变量，spark会向Executor端每个task都发送一个此变量的副本，
47. 如果此变量很大，就会占用大量的Executor节点的内存，
48. 利用广播变量，spark只会给一个Executor节点发送一个变量
49. */
50. val bc:BroadCast[T] = sc.broadcast(value:T)
51. rdd.mapPartitions(itPar=>{
52.     val v:T = bc.value
53.     ...
54. })
56. /*
57. 累加器：accumulate：只能 add 操作，常用于计数（累加器在Driver定义初始化，在Excutor端更新）
58.         1、定义在Driver端的一个变量，Cluster中每一个Task都会有一份Copy
59.         2、所有的Task都计算完成后，将所有Task中的Copy合并到驱动程序中的变量中
60. 非累加器：在所有Task中的都会是独立Copy，不会有合并
61. 自定义累加器：写一个类继承 AccumulatorV2[IN, OUT]
62. abstract class AccumulatorV2[IN, OUT] extends Serializable {
63.   // Returns if this accumulator is zero value or not
64.   def isZero: Boolean
66.   //  Creates a new copy of this accumulator, which is zero value
67.   def copyAndReset(): AccumulatorV2[IN, OUT] = {...}
69.   // Creates a new copy of this accumulator.
70.   def copy(): AccumulatorV2[IN, OUT]
72.   // Resets this accumulator, which is zero value.
73.   def reset(): Unit
75.   // 添加：Takes the inputs and accumulates.
76.   def add(v: IN): Unit
78.   // 合并：Merges another same-type accumulator and update its state.
79.   def merge(other: AccumulatorV2[IN, OUT]): Unit
81.   // 值列表：Defines the current value of this accumulator
82.   def value: OUT
83. }
84. */
85. val accLong: LongAccumulator = sc.longAccumulator("longAcc")
86. val accDouble: DoubleAccumulator = sc.doubleAccumulator("doubleAcc")
87. rdd.mapPartitions(itPar=>{
88.     ...
89.     accLong.add(v:Long)
90.     accDouble.add(v:Double)
91.     ...
92. })
93. accXxx.reset()
94. val isZero:Boolean = accXxx.isZero
95. val num:Long|Double = accXxx.value|sum|count|avg
97. /*
98. 【分区控制】
99.         【缩减分区节省资源】 或 【扩大分区提高并行度】
100.          coalesce(numPartitions:Int, shuffle:Boolean):
101.                 缩小分区
102.                     存在过多的小任务的时候收缩合并分区，减少分区的个数，减少任务调度成本
103.                     默认情况下，不会对数据重组，比如：3个合成2个，采用 {1+2}，{3}，容易导致数据倾斜
104.                     若需数据均衡，则将 shuffle 参数设置为 true 即可
105.             扩大分区
106.                     若需要扩大分区，shuffle 参数必须设置为 true
107.                     若将2个分区拆分成3个，必须打乱重新分区，否则数据还是在两个分区，{1}，{2}，{空}
108.                     repartition(numPartitions:Int) <=> coalesce(numPartitions,true)
109. */
110. val rdd: RDD[String] = rdd.coalesce(numPartitions:Int, shuffle:Boolean)
111. val rdd: RDD[String] = rdd.repartition(numPartitions:Int) // ✔

复制代码

阶段分别 DAG

1. /*
2. 【为什么要划分阶段】
3.         1、基于数据的分区，本着传递计算的性能远高于传递数据，所以数据本地化是提升性能的重要途径之一
4.         2、一组串行的算子，无需 Shuffle，基于数据本地化可以作为一个独立的阶段连续执行
5.         3、经过一组串行算子计算，遇到 Shuffle 操作，默认情况下 Shuffle 不会改变分区数量，但会因为numPartitions:Int, partitioner:Partitioner 等参数重新分配，过程数据会【写盘供子RDD拉取(类MapReduce)】
6. */
8. /*
9. Driver程序提交后
10. 1、Spark调度器将所有的RDD看成是一个Stage
11. 2、然后对此Stage进行逆向回溯，遇到Shuffle就断开，形成一个新的Stage
12. 3、遇到窄依赖，则归并到同一个Stage(TaskSet)
13. 4、等到所有的步骤回溯完成，便生成一个DAG图
15. RDD依赖关系
16.         Lineage：血统、遗传
17.         RDD最重要的特性之一，保存了RDD的依赖关系
18.         RDD实现了基于Lineage的容错机制
19.         依赖关系 org.apache.spark.Dependency
20.         窄依赖 NarrowDependency
21.                 1V1 OneToOneDependency
22.                 1VN RangeDependency
23.         宽依赖 ShuffleDependency
24.     当RDD分区丢失时
25.         Spark会对数据进行重新计算，对于窄依赖只需重新计算一次子RDD的父RDD分区
26.         若配合持久化更佳：cache,persist,checkpoint
27. */

复制代码

算子宽窄依赖分别

1. // 窄依赖
2. rdd.dependencies
3.         map
4.         flatMap
5.         mapPartitions ✔
6.         mapPartitionsWithIndex
7.         glom
8.         filter
9.         distinct
10.         intersection  ✔
11.         sample ✔
12.         union ✔
13.         subtract ✔
14.         zip...
15.         cogroup
16. // 宽依赖
17. ShuffledRDD extends RDD
18.         sortBy
19.         sortByKey
20.         partitionBy
21.         repartition
23. // 不一定
24. /*
25. reduceByKey(【partitioner: Partitioner】, func: (V, V) => V)
26.         若使用的是带 partitioner 的重载且 Partitioner 和父RDD的 Partitioner一致
27.         则为窄依赖RDD，否则为宽依赖ShuffledRDD
28. */

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。