1. 程式人生 > >Spark開發的完整基礎_歡樂的馬小紀

Spark開發的完整基礎_歡樂的馬小紀

 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)(_+_, _+_)

rdd1.aggregate(0)(math.max(_, _), _ + _)

###5和1比, 得5再和234比得5 --> 5和6789比,得9 --> 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("=")(_ + _, _ + _)

 

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)

 

val rdd4 = sc.parallelize(List("12","23","345",""),2)

rdd4.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)

兩個分割槽

1.("","12","23")->("0","23")->("1")

2. ("","345","")  ->("0","")  ->("0")

 

val rdd5 = sc.parallelize(List("12","23","","345"),2)

rdd5.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)

兩個分割槽

1.("","12","23")->("0","23")->("1")

2. ("","","345")  ->("1","")  ->("1")

-------------------------------------------------------------------------------------------

-------------------------------------------------------------------------------------------

aggregateByKey  和 reduceByKey基本一樣,區別是它同於combiner

 

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)(math.max(_, _), _ + _).collect

pairRDD.aggregateByKey(100)(math.max(_, _), _ + _).collect

-------------------------------------------------------------------------------------------

-------------------------------------------------------------------------------------------

checkpoint

sc.setCheckpointDir("hdfs://node-1.itcast.cn:9000/ck")

val rdd = sc.textFile("hdfs://node-1.itcast.cn:9000/wc").flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_)

rdd.checkpoint

rdd.isCheckpointed

rdd.count

rdd.isCheckpointed

rdd.getCheckpointFile

-------------------------------------------------------------------------------------------

-------------------------------------------------------------------------------------------

coalesce, repartition

val rdd1 = sc.parallelize(1 to 10, 10)

val rdd2 = rdd1.coalesce(2, false)

rdd2.partitions.length

coalesce等同於repartition,第二個引數指的是否進行shuffle,

repartition方法就是呼叫coalesce方法,-----repartition(a)等同於coalesce(a,true)

-------------------------------------------------------------------------------------------

-------------------------------------------------------------------------------------------

collectAsMap : Map(b -> 2, a -> 1)

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))

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

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)

val rdd6 = rdd5.zip(rdd4)

val rdd7 = rdd6.combineByKey(List(_), (x: List[String], y: String) => x :+ y, (m: List[String], n: List[String]) => m ++ n)

-------------------------------------------------------------------------------------------

-------------------------------------------------------------------------------------------

countByKey 

val rdd1 = sc.parallelize(List(("a", 1), ("b", 2), ("b", 2), ("c", 2), ("c", 1)))

rdd1.countByKey--------------Map(a->1,b->2,c->2)

rdd1.countByValue------------Map(("a", 1)->1,("b", 2)->2,("c", 2)->1,("c", 1)->1)

-------------------------------------------------------------------------------------------

-------------------------------------------------------------------------------------------

filterByRange

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))

val rdd3 = sc.parallelize(List(("a", "1 2"), ("b", "3 4")))

val rdd4 = rdd3.flatMapValues(_.split(" "))--------------------------Array((a,1), (a,2), (b,3), (b,4))

rdd4.collect

-------------------------------------------------------------------------------------------

-------------------------------------------------------------------------------------------

foldByKey 

 

val rdd1 = sc.parallelize(List("dog", "wolf", "cat", "bear"), 2)

val rdd2 = rdd1.map(x => (x.length, x))

val rdd3 = rdd2.foldByKey("")(_+_)---------------((3,dogcat),(4,wolf,bear))

 

val rdd = sc.textFile("hdfs://node-1.itcast.cn:9000/wc").flatMap(_.split(" ")).map((_, 1))

rdd.foldByKey(0)(_+_)

-------------------------------------------------------------------------------------------

-------------------------------------------------------------------------------------------

foreachPartition  action操作,雖然不能返回RDD,但是可以在裡面對分割槽進行操作

val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 3)

rdd1.foreachPartition(x => println(x.reduce(_ + _)))  

-------------------------------------------------------------------------------------------

-------------------------------------------------------------------------------------------

keyBy : 以傳入的引數做key

val rdd1 = sc.parallelize(List("dog", "salmon", "salmon", "rat", "elephant"), 3)

val rdd2 = rdd1.keyBy(_.length)

val rdd2 = rdd1.keyBy(_(0))

rdd2.collect

-------------------------------------------------------------------------------------------

-------------------------------------------------------------------------------------------

keys values

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

-------------------------------------------------------------------------------------------

-------------------------------------------------------------------------------------------