定義不帶參數也不帶返回值的函數（def :定義函數的關鍵字 printz:方法名稱）
scala> def printz = print("scala hello")
定義帶參數也帶返回值的函數（這種函數在定義時也可以不帶返回值的類型，scala會自動推算出。建議還是帶上）
scala> def minNum(x:Int,y:Int):Int = if(x>y) x else y //:Int 是該函數的返回值類型 minNum: (x: Int, y: Int)Int
調用函數
scala> minNum(2,3) res10: Int = 3
創建一個map
scala> val colors = Map("red" -> "#FF0000", "azure" -> "#F0FFFF") colors: scala.collection.immutable.Map[String,String] = Map(red -> #FF0000, azur e -> #F0FFFF)
將map中的每一個元素中的第一個元組與第二個元組交換位置
'_'：表示map的元組，
x._2：map中每一個元素的第二個元組（#FF0000,"#F0FFFF"）
scala> colors.map(x=>(x._2,x._1)) res0: scala.collection.immutable.Map[String,String] = Map(#FF0000 -> red, #F0FFF F -> azure)
union(合並rdd)
scala> var rdd1=sc.parallelize(List(1,2,3,4))//創建rdd1 rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[53] at parallelize at <console>:24 scala> var rdd2=sc.parallelize(List(5,6,7,8))//創建rdd2 rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[54] at parallelize at <console>:24 scala> var unionRes=rdd1 union rdd2 //合並rdd unionRes: org.apache.spark.rdd.RDD[Int] = UnionRDD[55] at union at <console>:28 scala> unionRes.collect //查合並后的rdd res19: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8)
join 類似於sql中的內連，左外鏈接，右外鏈接
創建rdd
scala> var rdd1=sc.parallelize(List((1,"spark"),(2,"hadoop"),(3,"scala"),(4,"java"))) //創建rdd1 rdd1: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[58] at parallelize at <console>:24 scala> var rdd2=sc.parallelize(List((1,"30k"),(2,"28k"),(3,"5k"),(5,"10k"))) //創建rddd2 rdd2: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[59] at parallelize at <console>:24 scala> var joinRes=rdd1 join rdd2 //join rdd joinRes: org.apache.spark.rdd.RDD[(Int, (String, String))] = MapPartitionsRDD[62] at join at <console>:28 scala> joinRes.collect //查看join后的結果，可以看出join會把兩個rdd有相同key的元素的值進行合並 res20: Array[(Int, (String, String))] = Array((1,(spark,30k)), (3,(scala,5k)), (2,(hadoop,28k)))
因為第4和第5沒有匹配項，所以不顯示
leftOuterJoin（左外鏈接）
//這個是以左邊的rdd為主，右邊為輔。右邊與左邊沒有匹配項，則會顯示左邊的值，所以這里key為4的也會顯示 scala> var leftJoinRes=rdd1.leftOuterJoin(rdd2) leftJoinRes: org.apache.spark.rdd.RDD[(Int, (String, Option[String]))] = MapPartitionsRDD[65] at leftOuterJoin at <console>:28 scala> leftJoinRes.collect res21: Array[(Int, (String, Option[String]))] = Array((4,(java,None)), (1,(spark,Some(30k))), (3,(scala,Some(5k))), (2,(hadoop,Some(28k))))
rightOuterJoin(右外鏈接)
//這個是以右邊邊的rdd為主，左邊邊為輔。右邊與左邊沒有匹配項，則只顯示右邊,所以這里key為5的也會顯示 scala> var rightJoinRes=rdd1.rightOuterJoin(rdd2) rightJoinRes: org.apache.spark.rdd.RDD[(Int, (Option[String], String))] = MapPartitionsRDD[68] at rightOuterJoin at <console>:28 scala> rightJoinRes.collect res22: Array[(Int, (Option[String], String))] = Array((1,(Some(spark),30k)), (3,(Some(scala),5k)), (5,(None,10k)), (2,(Some(hadoop),28k)))
groupByKey
//(4,1), (7,4), (6,3), (2,2), (3,3), (1,3)
scala> f1.flatMap(x=>x.split("-")).map((_,1)).groupByKey.collect res28: Array[(String, Iterable[Int])] = Array((4,CompactBuffer(1)), (7,CompactBuffer(1, 1, 1, 1)), (6,CompactBuffer(1, 1, 1)), (2,CompactBuffer(1, 1)), (3,CompactBuffer(1, 1, 1)), (1,CompactBuffer(1, 1, 1)))
reduce
scala> var rdd1=sc.parallelize(List(1,2,3,4,5)) rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[80] at parallelize at <console>:24 scala> rdd1.reduce(_+_) res32: Int = 15
reduce(_+_):每個元組相加 1+2+3+4+5=15
他是每次相加兩個元組然后產生新的rdd與下一位元組相加
1+2=3
3+3=6
6+4=10
10+5=15
lookup
scala> var rdd1=sc.parallelize(List((1,"a"),(2,"b"),(3,"c"))) rdd1: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[81] at parallelize at <console>:24 scala> rdd1.lookup(1) res34: Seq[String] = WrappedArray(a)
lookup函數對<key,value>型的rdd操作，返回指定key對應的元素形成的seq，這個函數的優點
如果這個rdd包含分區器，那么只掃描對應key所在的分區，然后返回對應key的元素形成的seq；如果這個rdd沒有分區器，則對這個rdd進行全盤掃描，然后返回對應key的元素形成的seq
map和flatMap
map函數是以一行數據為一個元素
Array[Student] = Array(Student(zhangxs,24,chenxy), Student(wangYr,21,teacher), Student(wangx,26,teacher))
map是對RDD中的每個元素都執行一個指定的函數來產生一個新的RDD。任何原RDD中的元素在新RDD中都有且只有一個元素與之對應。
scala> val a = sc.parallelize(1 to 9, 3) scala> val b = a.map(x => x*2) scala> a.collect res10: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9) scala> b.collect res11: Array[Int] = Array(2, 4, 6, 8, 10, 12, 14, 16, 18)
上述例子中把原RDD中每個元素都乘以2來產生一個新的RDD。
flatMap是一個文件的數據為一個元素
res9: Array[String] = Array(zhangxs, 24, chenxy, wangYr, 21, teacher, wangx, 26, teacher)
mapPartitions
mapPartitions是map的一個變種。map的輸入函數是應用於RDD中每個元素，而mapPartitions的輸入函數是應用於每個分區，也就是把每個分區中的內容作為整體來處理的。
它的函數定義為：
def mapPartitions[U: ClassTag](f: Iterator[T] => Iterator[U], preservesPartitioning: Boolean = false): RDD[U]
f即為輸入函數，它處理每個分區里面的內容。每個分區中的內容將以Iterator[T]傳遞給輸入函數f，f的輸出結果是Iterator[U]。最終的RDD由所有分區經過輸入函數處理后的結果合並起來的。
比如
scala> val a = sc.parallelize(1 to 9, 3) scala> def myfunc[T](iter: Iterator[T]) : Iterator[(T, T)] = { var res = List[(T, T)]() var pre = iter.next while (iter.hasNext) { val cur = iter.next; res .::= (pre, cur) pre = cur; } res.iterator } scala> a.mapPartitions(myfunc).collect res0: Array[(Int, Int)] = Array((2,3), (1,2), (5,6), (4,5), (8,9), (7,8))
上述例子中的函數myfunc是把分區中一個元素和它的下一個元素組成一個Tuple。因為分區中最后一個元素沒有下一個元素了，所以(3,4)和(6,7)不在結果中。
mapPartitions還有些變種，比如mapPartitionsWithContext，它能把處理過程中的一些狀態信息傳遞給用戶指定的輸入函數。還有mapPartitionsWithIndex，它能把分區的index傳遞給用戶指定的輸入函數。
mapValues
mapValues顧名思義就是輸入函數應用於RDD中Kev-Value的Value，原RDD中的Key保持不變，與新的Value一起組成新的RDD中的元素。因此，該函數只適用於元素為KV對的RDD。
scala> val a = sc.parallelize(List("dog", "tiger", "lion", "cat", "panther", " eagle"), 2) scala> val b = a.map(x => (x.length, x)) //key：元組的長度 value：元組的值 scala> b.mapValues("x" + _ + "x").collect //“_”就是每個元組的value，在每個value的前后加"x" res5: Array[(Int, String)] = Array((3,xdogx), (5,xtigerx), (4,xlionx),(3,xcatx), (7,xpantherx), (5,xeaglex))
mapWith
mapWith是map的另外一個變種，map只需要一個輸入函數，而mapWith有兩個輸入函數。它的定義如下：
def mapWith[A: ClassTag, U: ](constructA: Int => A, preservesPartitioning: Boolean = false)(f: (T, A) => U): RDD[U]
- 第一個函數constructA是把RDD的partition index（index從0開始）作為輸入，輸出為新類型A；
- 第二個函數f是把二元組(T, A)作為輸入（其中T為原RDD中的元素，A為第一個函數的輸出），輸出類型為U。
比如把partition index 乘以10，然后加上2作為新的RDD的元素。
val x = sc.parallelize(List(1,2,3,4,5,6,7,8,9,10), 3) x.mapWith(a => a * 10)((a, b) => (b + 2)).collect res4: Array[Int] = Array(2, 2, 2, 12, 12, 12, 22, 22, 22, 22)