spark的RDD高階運算元
map是對每個元素操作, mapPartitions是對其中的每個partition操作
mapPartitionsWithIndex : 把每個partition中的分割槽號和對應的值拿出來, 看原始碼
val func = (index: Int, iter: Iterator[(Int)]) => {
iter.toList.map(x => “[partID:” + index + ", val: " + x + “]”).iterator
}
val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)
rdd1.mapPartitionsWithIndex(func).collect
aggregate(初始值+目標值)+(第二個分割槽初始值+目標值)…
#先得到分割槽的加和再彙總時使用
def func1(index: Int, iter: Iterator[(Int)]) : Iterator[String] = {
iter.toList.map(x => “[partID:” + index + ", val: " + x + “]”).iterator
}
val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)
rdd1.mapPartitionsWithIndex(func1).collect
###是action操作, 第一個引數是初始值, 二:是2個函式[每個函式都是2個引數(第一個引數:先對個個分割槽進行合併, 第二個:對個個分割槽合併後的結果再進行合併), 輸出一個引數]
###0 + (0+1+2+3+4 + 0+5+6+7+8+9)
rdd1.aggregate(0)(+
#(0+4)+(0+9)
rdd1.aggregate(0)(math.max(_, _), _ + )
###5和1比, 得5再和234比得5 --> 5和6789比,得9 --> (5+5) + (5+9)
rdd1.aggregate(5)(math.max(, _), _ + _)
val rdd2 = sc.parallelize(List(“a”,“b”,“c”,“d”,“e”,“f”),2)
def func2(index: Int, iter: Iterator[(String)]) : Iterator[String] = {
iter.toList.map(x => “[partID:” + index + “, val: " + x + “]”).iterator
}
rdd2.aggregate(”")(_ + _, _ + )
rdd2.aggregate("|")(
rdd2.aggregate("=")( + _, _ + _)
rdd3.mapPartitionsWithIndex(func2).collect
val rdd3 = sc.parallelize(List(“12”,“23”,“345”,“4567”),2)
rdd3.aggregate("")((x,y) => math.max(x.length, y.length).toString, (x,y) => x + y)
#結果為24或42因為平行計算
rdd3.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)
#結果為11
rdd4.mapPartitionsWithIndex(func2).collect
val rdd4 = sc.parallelize(List(“12”,“23”,“345”,""),2)
#"“的length toString為0,0的長度再和x的length相比得到最小長度為1
rdd4.aggregate(”")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)
rdd4.aggregate("")((x,y) => math.max(x.length, y.length).toString, (x,y) => x + y)
val rdd5 = sc.parallelize(List(“12”,“23”,"",“345”),2)
rdd5.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)
aggregateByKey
#把Key相同的進行加和
val pairRDD = sc.parallelize(List( (“cat”,2), (“cat”, 5), (“mouse”, 4),(“cat”, 12), (“dog”, 12), (“mouse”, 2)), 2)
def func2(index: Int, iter: Iterator[(String, Int)]) : Iterator[String] = {
iter.toList.map(x => “[partID:” + index + ", val: " + x + “]”).iterator
}
pairRDD.mapPartitionsWithIndex(func2).collect
pairRDD.aggregateByKey(0)(_ + _, _ + ).collect
pairRDD.aggregateByKey(0)(math.max(, _), _ + ).collect
pairRDD.aggregateByKey(100)(math.max(, _), _ + _).collect
checkpoint (檢查點)(轉換動作)
//設定共享儲存位置,如果rdd找不到,程式自己先檢查cache然後檢查checkpoint
//一般把一些比較重要東西checkpoint (如計算出的結果也就是reduceByKey執行後,強烈建議使用前將資料持久化到記憶體)
//必須在action之前執行
//也就是rdd先cache,再checkpoint ,再collect
//做非常複雜的運算時候資料有可能丟失時使用(即時cache也有可能丟失)
sc.setCheckpointDir(“hdfs://node-1.itcast.cn:9000/ck”)
sc.setCheckpointDir(“file:///root/ck”)
val rdd = sc.textFile(“hdfs://node-1.itcast.cn:9000/wc”).flatMap(.split(" ")).map((, 1)).reduceByKey(+)
val rdd = sc.textFile(“file:///root//wc.txt”).flatMap(.split(" ")).map((, 1)).reduceByKey(+)
rdd.checkpoint
rdd.isCheckpointed
rdd.count
rdd.isCheckpointed
rdd.getCheckpointFile
coalesce, repartition
#合併為N個分割槽
val rdd1 = sc.parallelize(1 to 10, 10)
val rdd2 = rdd1.coalesce(2, false)
rdd2.partitions.length
collectAsMap : Map(b -> 2, a -> 1)
#把一個collect專為map
val rdd = sc.parallelize(List((“a”, 1), (“b”, 2)))
rdd.collectAsMap
combineByKey : 和reduceByKey是相同的效果
###第一個引數x:原封不動取出來, 第二個引數:是函式, 區域性運算, 第三個:是函式, 對區域性運算後的結果再做運算
###每個分割槽中每個key中value中的第一個值, (hello,1)(hello,1)(good,1)–>(hello(1,1),good(1))–>x就相當於hello的第一個1, good中的1
val rdd1 = sc.textFile(“hdfs://master:9000/wordcount/input/”).flatMap(.split(" ")).map((, 1))
#a和b先區域性求和(分割槽內部求和),m和n在分割槽外部求和
val rdd2 = rdd1.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
rdd1.collect
rdd2.collect
###當input下有3個檔案時(有3個block塊, 不是有3個檔案就有3個block, ), 每個會多加3個10
#a+b是計算各個分割槽中相同key的值,後面mn是計算所有分割槽中相同key的值
val rdd3 = rdd1.combineByKey(x => x + 10, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
rdd3.collect
val rdd4 = sc.parallelize(List(“dog”,“cat”,“gnu”,“salmon”,“rabbit”,“turkey”,“wolf”,“bear”,“bee”), 3)
val rdd5 = sc.parallelize(List(1,1,2,2,2,1,2,2,2), 3)
#zip把兩個rdd合併為一個array
val rdd6 = rdd5.zip(rdd4)
#x :+ y在一個list後面追加元素,這一步是在每個分割槽中計算得到list,後面mn是合併各個分割槽的list
val rdd7 = rdd6.combineByKey(List(_), (x: List[String], y: String) => x :+ y, (m: List[String], n: List[String]) => m ++ n)
countByKey (統計每個key的數量)(countByValue統計每個kv的數量)
val rdd1 = sc.parallelize(List((“a”, 1), (“b”, 2), (“b”, 2), (“c”, 2), (“c”, 1)))
rdd1.countByKey
rdd1.countByValue
filterByRange(把b和d中間的字母取出來)
val rdd1 = sc.parallelize(List((“e”, 5), (“c”, 3), (“d”, 4), (“c”, 2), (“a”, 1)))
val rdd2 = rdd1.filterByRange(“b”, “d”)
rdd2.collect
flatMapValues : Array((a,1), (a,2), (b,3), (b,4))(把一個array做成多個array)
val rdd3 = sc.parallelize(List((“a”, “1 2”), (“b”, “3 4”)))
val rdd4 = rdd3.flatMapValues(_.split(" "))
rdd4.collect
foldByKey(該函式用於RDD[K,V]根據K將V做摺疊、合併處理，其中的引數zeroValue表示先根據對映函式將zeroValue應用於V,進行初始化V,再將對映函式應用於初始化後的V.)
val rdd1 = sc.parallelize(List(“dog”, “wolf”, “cat”, “bear”), 2)
val rdd2 = rdd1.map(x => (x.length, x))
val rdd3 = rdd2.foldByKey("")(+)
val rdd = sc.textFile(“hdfs://node-1.itcast.cn:9000/wc”).flatMap(.split(" ")).map((, 1))
val rdd = sc.textFile(“file:///root//wc.txt”).flatMap(.split(" ")).map((, 1)).reduceByKey(+)
rdd.foldByKey(0)(+)
foreachPartition(把每個分割槽拿出來操作,不會產生新的rdd)
val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 3)
rdd1.foreachPartition(x => println(x.reduce(_ + _)))
#結果為 6 15 24
keyBy : 以傳入的引數做key
val rdd1 = sc.parallelize(List(“dog”, “salmon”, “salmon”, “rat”, “elephant”), 3)
val rdd2 = rdd1.keyBy(_.length)
rdd2.collect
keys values(得到資料的key和value)
val rdd1 = sc.parallelize(List(“dog”, “tiger”, “lion”, “cat”, “panther”, “eagle”), 2)
val rdd2 = rdd1.map(x => (x.length, x))
rdd2.keys.collect
rdd2.values.collect
mapPartitions
(與map方法類似，map是對rdd中的每一個元素進行操作，而mapPartitions(foreachPartition)則是對rdd中的每個分割槽的迭代器進行操作。如果在map過程中需要頻繁建立額外的物件(例如將rdd中的資料通過jdbc寫入資料庫,map需要為每個元素建立一個連結而mapPartition為每個partition建立一個連結),則mapPartitions效率比map高的多)
val a = sc.parallelize(1 to 9, 3)
def doubleFunc(iter: Iterator[Int]) : Iterator[(Int,Int)] = {
var res = List(Int,Int)
while (iter.hasNext)
{
val cur = iter.next;
res .::= (cur,cur*2)
}
res.iterator
}
val result = a.mapPartitions(doubleFunc)
println(result.collect().mkString)