Spark的transformation 和 action的操作學習筆記

一、spark的transformation 和 action區別

Spark有一些基本的transformation 和 action的操作，其中transformation形成各類型的RDD，action不形成RDD，而是對RDD進行累加、合并、保存操作。

二、transformation 有哪些

transformation有map、filter、flatMap（與map不一樣）、Sample、groupByKey、ReduceByKey、Union、Join、cogroup、crossProduct、mapValues、sort、partitionBy，共13種。還有sortByKey呢？

1、map：

val rdd = sc.parallelize(List(1,2,3,4,5,6))  

val mapRdd = rdd.map(_*2)  //這是典型的函數式編程

mapRdd.collect()  //上面的map是transformation，到了這里的collect才開始執行，是action，返回一個Array    Array(2,4,6,8,10,12)

map(x=>(x,1))，將map(x)這樣的，映射成map(x,1)這樣的，一般用于對Key進行計數

2、filter

過濾，選擇函數，

val filterRdd = mapRdd.filter(_ > 5)

filterRdd.collect() //返回所有大于5的數據的一個Array， Array(6,8,10,12)

3、flatmap加上reduceBykey

val wordcount = rdd.flatMap(_.split(' ')).map((_, 1)).reduceByKey(_+_)  //把每一行進行根據空格分割，然后flatMap會把多個list合并成一個list，最后把每個元素變成一個元組

//然后把具有相同key的元素的value進行相加操作，參考上面圖片中的函數定義，針對reduceByKey，傳入的函數是對value進行操作的。

wordcount.saveAsTextFile("/xxx/ss/aa")   //把結果存入文件系統

wordcount.collect //可以得到一個數組

4、groupByKey

對文件按照空格進行分割后，按照單詞進行groupByKey分組

val wordcount=rdd.flatMap(_.split(' ')).map(_.1)).groupByKey

使用collect查看一下結果

wordcount.collect

5、Union

2個合并成1個

val rdd1 = sc.parallelize(List(('a',1),(‘a’, 2)))

val rdd2 = sc.parallelize(List(('b',1),(‘b’, 2)))

val result_union = rdd1 union rdd2 //結果是把兩個list合并成一個，List(('a',1),(‘a’, 2),('b',1),(‘b’, 2))

6、Join

笛卡爾積的干活，小組循環賽

val rdd1 = sc.parallelize(List(('a',1),(‘a’, 2), ('b', 3)))

val rdd2 = sc.parallelize(List(('a',4),(‘b’, 5)))

val result_union = rdd1 join rdd2 //結果是把兩個list做笛卡爾積，Array(('a', (1,4), ('a', (2,4), ('b', (3, 5)))

7、sortByKey

排序，非常好用的哈

val wordcount = rdd.flatMap(_split(' ')).map(_,1).reduceByKey(_+_).map(x => (x._2, x._1)).sortByKey(false).map(x => (x._2, x._1))

//其實完成了一個sort by value的過程， sortByKey(false)，表示倒序排列

三、action有哪些

action有count、collect、reduce、lookup、save5種。

1、count

計算rdd的個數

val rdd = sc.textFile("/xxx/sss/ee")

rdd.count //計算行數

rdd.cache   //可以把rdd保留在內存里面

rdd.count //計算行數，但是因為上面進行了cache，這里速度會很快

2、collect

collect函數可以提取出所有rdd里的數據項

val rdd1=sc.parallelize(List(('a',1),('b',1)))

val rdd2=sc.parallelize(List(('c',1),('d',1)))

val result=rdd1 union rdd2

使用collect操作查看一下執行結果

3、reduce

map、reduce是hadoop的2個核心，map是映射，reduce是精簡

val rdd = sc.parallelize(List(1,2,3,4))

rdd.reduce(_+_) //reduce是一個action，這里的結果是10

4、lookup

查找的干活

val rdd = sc.parallelize(List(('a',1),(‘a’, 2),('b',1),(‘b’, 2))

rdd.lookup("a") //返回一個seq， (1, 2) 是把a對應的所有元素的value提出來組成一個seq

5、save

查詢搜索結果排名第 1 點擊次序排在第 2 的數據

val rdd1 = sc.textFile("hdfs://192.168.0.10:9000/input/SogouQ2.txt").map(_.split("\t"))   //長度為6錯誤，好像日志不標準，有的為6，有的不是  .filter(_.length==6)

rdd1.count()

val rdd2=rdd1.filter(_(3).toInt==1).filter(_(4).toInt==2).count()

rdd2.saveAsTextFile("hdfs://192.168.0.10:9000/output/sogou1111/")