廣東省建設(shè)廳網(wǎng)站優(yōu)化師培訓(xùn)機(jī)構(gòu)
重分區(qū)函數(shù)
如何對(duì)RDD中分區(qū)數(shù)目進(jìn)行調(diào)整（增加分區(qū)或減少分區(qū)），在RDD函數(shù)中主要有如下三個(gè)函數(shù)。
1）、增加分區(qū)函數(shù)
函數(shù)名稱：repartition，此函數(shù)使用的謹(jǐn)慎，會(huì)產(chǎn)生Shuffle。
2）、減少分區(qū)函數(shù)
函數(shù)名稱：coalesce，此函數(shù)不會(huì)產(chǎn)生Shuffle，當(dāng)且僅當(dāng)降低RDD分區(qū)數(shù)目。
比如RDD的分區(qū)數(shù)目為10個(gè)分區(qū)，此時(shí)調(diào)用rdd.coalesce(12)，不會(huì)對(duì)RDD進(jìn)行任何操作。
3）、調(diào)整分區(qū)函數(shù)
在PairRDDFunctions（此類專門針對(duì)RDD中數(shù)據(jù)類型為KeyValue對(duì)提供函數(shù)）工具類中
partitionBy函數(shù)：
范例演示代碼，適當(dāng)使用函數(shù)調(diào)整RDD分區(qū)數(shù)目：
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/**
* RDD中分區(qū)函數(shù)，調(diào)整RDD分區(qū)數(shù)目，可以增加分區(qū)和減少分區(qū)
*/
object SparkPartitionTest {
def main(args: Array[String]): Unit = {
// 創(chuàng)建應(yīng)用程序入口SparkContext實(shí)例對(duì)象
val sc: SparkContext = {
// 1.a 創(chuàng)建SparkConf對(duì)象，設(shè)置應(yīng)用的配置信息
val sparkConf: SparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[2]")
// 1.b 傳遞SparkConf對(duì)象，構(gòu)建Context實(shí)例
new SparkContext(sparkConf)
}
sc.setLogLevel("WARN")
// 讀取本地文件系統(tǒng)文本文件數(shù)據(jù)
val datasRDD: RDD[String] = sc.textFile("datas/wordcount/wordcount.data", minPartitions = 2)
// TODO: 增加RDD分區(qū)數(shù)
val etlRDD: RDD[String] = datasRDD.repartition(3)
println(s"EtlRDD 分區(qū)數(shù)目 = ${etlRDD.getNumPartitions}")
// 詞頻統(tǒng)計(jì)
val resultRDD: RDD[(String, Int)] = etlRDD
// 數(shù)據(jù)分析，考慮過濾臟數(shù)據(jù)
.filter(line => null != line && line.trim.length > 0)
// 分割單詞，注意去除左右空格
.flatMap(line => line.trim.split("\\s+"))
// 轉(zhuǎn)換為二元組，表示單詞出現(xiàn)一次
.mapPartitions{iter =>
iter.map(word => (word, 1))
}
// 分組聚合，按照Key單詞
.reduceByKey((tmp, item) => tmp + item)
// 輸出結(jié)果RDD
resultRDD
// TODO: 對(duì)結(jié)果RDD降低分區(qū)數(shù)目
.coalesce(1)
.foreachPartition(iter => iter.foreach(println))
// 應(yīng)用程序運(yùn)行結(jié)束，關(guān)閉資源
sc.stop()
}
}
在實(shí)際開發(fā)中，什么時(shí)候適當(dāng)調(diào)整RDD的分區(qū)數(shù)目呢？讓程序性能更好好呢？？？？
第一點(diǎn)：增加分區(qū)數(shù)目
- 當(dāng)處理的數(shù)據(jù)很多的時(shí)候，可以考慮增加RDD的分區(qū)數(shù)目
第二點(diǎn)：減少分區(qū)數(shù)目
- 其一：當(dāng)對(duì)RDD數(shù)據(jù)進(jìn)行過濾操作（filter函數(shù)）后，考慮是否降低RDD分區(qū)數(shù)目
- 其二：當(dāng)對(duì)結(jié)果RDD存儲(chǔ)到外部系統(tǒng)
聚合函數(shù)
在數(shù)據(jù)分析領(lǐng)域中，對(duì)數(shù)據(jù)聚合操作是最為關(guān)鍵的，在Spark框架中各個(gè)模塊使用時(shí)，主要就是其中聚合函數(shù)的使用。
集合中聚合函數(shù)
回顧列表List中reduce聚合函數(shù)核心概念：聚合的時(shí)候，往往需要聚合中間臨時(shí)變量。查看列表List中聚合函數(shù)reduce和fold源碼如下：
通過代碼，看看列表List中聚合函數(shù)使用：
運(yùn)行截圖如下所示：
fold聚合函數(shù)，比reduce聚合函數(shù)，多提供一個(gè)可以初始化聚合中間臨時(shí)變量的值參數(shù)：
聚合操作時(shí)，往往聚合過程中需要中間臨時(shí)變量（到底時(shí)幾個(gè)變量，具體業(yè)務(wù)而定），如下案例：
RDD 中聚合函數(shù)
在RDD中提供類似列表List中聚合函數(shù)reduce和fold，查看如下：
案例演示：求列表List中元素之和，RDD中分區(qū)數(shù)目為2，核心業(yè)務(wù)代碼如下：
運(yùn)行原理分析：
使用RDD中fold聚合函數(shù)：
查看RDD中高級(jí)聚合函數(shù)aggregate，函數(shù)聲明如下：
業(yè)務(wù)需求：使用aggregate函數(shù)實(shí)現(xiàn)RDD中最大的兩個(gè)數(shù)據(jù)，分析如下：
核心業(yè)務(wù)代碼如下：
運(yùn)行結(jié)果原理剖析示意圖：
上述完整范例演示代碼：
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext, TaskContext}
import scala.collection.mutable
import scala.collection.mutable.ListBuffer
/**
* RDD中聚合函數(shù)：reduce、aggregate函數(shù)
*/
object SparkAggTest {
def main(args: Array[String]): Unit = {
// 創(chuàng)建應(yīng)用程序入口SparkContext實(shí)例對(duì)象
val sc: SparkContext = {
// 1.a 創(chuàng)建SparkConf對(duì)象，設(shè)置應(yīng)用的配置信息
val sparkConf: SparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[2]")
// 1.b 傳遞SparkConf對(duì)象，構(gòu)建Context實(shí)例
new SparkContext(sparkConf)
}
sc.setLogLevel("WARN")
// 模擬數(shù)據(jù)，1 到 10 的列表，通過并行方式創(chuàng)建RDD
val datas = 1 to 10
val datasRDD: RDD[Int] = sc.parallelize(datas, numSlices = 2)
// 查看每個(gè)分區(qū)中的數(shù)據(jù)
datasRDD.foreachPartition{iter =>
println(s"p-${TaskContext.getPartitionId()}: ${iter.mkString(", ")}")
}
println("=========================================")
// 使用reduce函數(shù)聚合
val result: Int = datasRDD.reduce((tmp, item) => {
println(s"p-${TaskContext.getPartitionId()}: tmp = $tmp, item = $item")
tmp + item
})
println(result)
println("=========================================")
// 使用fold函數(shù)聚合
val result2: Int = datasRDD.fold(0)((tmp, item) => {
println(s"p-${TaskContext.getPartitionId()}: tmp = $tmp, item = $item")
tmp + item
})
println(result2)
println("=========================================")
// 使用aggregate函數(shù)獲取最大的兩個(gè)值
val top2: mutable.Seq[Int] = datasRDD.aggregate(new ListBuffer[Int]())(
// 分區(qū)內(nèi)聚合函數(shù)，每個(gè)分區(qū)內(nèi)數(shù)據(jù)如何聚合 seqOp: (U, T) => U,
(u, t) => {
println(s"p-${TaskContext.getPartitionId()}: u = $u, t = $t")
// 將元素加入到列表中
u += t //
// 降序排序
val top = u.sorted.takeRight(2)
// 返回
top
},
// 分區(qū)間聚合函數(shù)，每個(gè)分區(qū)聚合的結(jié)果如何聚合 combOp: (U, U) => U
(u1, u2) => {
println(s"p-${TaskContext.getPartitionId()}: u1 = $u1, u2 = $u2")
u1 ++= u2 // 將列表的數(shù)據(jù)合并，到u1中
//
u1.sorted.takeRight(2)
}
)
println(top2)
// 應(yīng)用程序運(yùn)行結(jié)束，關(guān)閉資源
sc.stop()
}
}
PairRDDFunctions 聚合函數(shù)
在Spark中有一個(gè)object對(duì)象PairRDDFunctions，主要針對(duì)RDD的數(shù)據(jù)類型是Key/Value對(duì)的數(shù)據(jù)提供函數(shù)，方便數(shù)據(jù)分析處理。比如使用過的函數(shù)：reduceByKey、groupByKey等。*ByKey函數(shù)：將相同Key的Value進(jìn)行聚合操作的，省去先分組再聚合。
第一類：分組函數(shù)groupByKey
第二類：分組聚合函數(shù)reduceByKey和foldByKey
但是reduceByKey和foldByKey聚合以后的結(jié)果數(shù)據(jù)類型與RDD中Value的數(shù)據(jù)類型是一樣的。
第三類：分組聚合函數(shù)aggregateByKey
在企業(yè)中如果對(duì)數(shù)據(jù)聚合使用，不能使用reduceByKey完成時(shí)，考慮使用aggregateByKey函數(shù)，基本上都能完成任意聚合功能。
演示范例代碼如下：
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/**
* RDD中聚合函數(shù)，針對(duì)RDD中數(shù)據(jù)類型Key/Value對(duì)：
* groupByKey
* reduceByKey/foldByKey
* aggregateByKey
* combineByKey
*/
object SparkAggByKeyTest {
def main(args: Array[String]): Unit = {
// 創(chuàng)建應(yīng)用程序入口SparkContext實(shí)例對(duì)象
val sc: SparkContext = {
// 1.a 創(chuàng)建SparkConf對(duì)象，設(shè)置應(yīng)用的配置信息
val sparkConf: SparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[2]")
// 1.b 傳遞SparkConf對(duì)象，構(gòu)建Context實(shí)例
new SparkContext(sparkConf)
}
sc.setLogLevel("WARN")
// 1、并行化集合創(chuàng)建RDD數(shù)據(jù)集
val linesSeq: Seq[String] = Seq(
"hadoop scala hive spark scala sql sql", //
"hadoop scala spark hdfs hive spark", //
"spark hdfs spark hdfs scala hive spark" //
)
val inputRDD: RDD[String] = sc.parallelize(linesSeq, numSlices = 2)
// 2、分割單詞，轉(zhuǎn)換為二元組
val wordsRDD: RDD[(String, Int)] = inputRDD
.flatMap(line => line.split("\\s+"))
.map(word => word -> 1)
// TODO: 先使用groupByKey函數(shù)分組，再使用map函數(shù)聚合
val wordsGroupRDD: RDD[(String, Iterable[Int])] = wordsRDD.groupByKey()
val resultRDD: RDD[(String, Int)] = wordsGroupRDD.map{ case (word, values) =>
val count: Int = values.sum
word -> count
}
println(resultRDD.collectAsMap())
// TODO: 直接使用reduceByKey或foldByKey分組聚合
val resultRDD2: RDD[(String, Int)] = wordsRDD.reduceByKey((tmp, item) => tmp + item)
println(resultRDD2.collectAsMap())
val resultRDD3 = wordsRDD.foldByKey(0)((tmp, item) => tmp + item)
println(resultRDD3.collectAsMap())
// TODO: 使用aggregateByKey聚合
/*
def aggregateByKey[U: ClassTag]
(zeroValue: U) // 聚合中間臨時(shí)變量初始值，類似fold函數(shù)zeroValue
(
seqOp: (U, V) => U, // 各個(gè)分區(qū)內(nèi)數(shù)據(jù)聚合操作函數(shù)
combOp: (U, U) => U // 分區(qū)間聚合結(jié)果的聚合操作函數(shù)
): RDD[(K, U)]
*/
val resultRDD4 = wordsRDD.aggregateByKey(0)(
(tmp: Int, item: Int) => {
tmp + item
},
(tmp: Int, result: Int) => {
tmp + result
}
)
println(resultRDD4.collectAsMap())
// 應(yīng)用程序運(yùn)行結(jié)束，關(guān)閉資源
Thread.sleep(1000000)
sc.stop()
}
}
面試題
RDD中g(shù)roupByKey和reduceByKey區(qū)別？？？
- reduceByKey函數(shù)：在一個(gè)(K,V)的RDD上調(diào)用，返回一個(gè)(K,V)的RDD，使用指定的reduce函數(shù)，將相同key的值聚合到一起，reduce任務(wù)的個(gè)數(shù)可以通過第二個(gè)可選的參數(shù)來設(shè)置。
- groupByKey函數(shù)：在一個(gè)(K,V)的RDD上調(diào)用，返回一個(gè)(K,V)的RDD，使用指定的函數(shù)，將相同key的值聚合到一起，與reduceByKey的區(qū)別是只生成一個(gè)sequence。
