一、spark的transformation 和 action區(qū)別

創(chuàng)新互聯(lián)IDC提供業(yè)務:綿陽主機托管,成都服務器租用,綿陽主機托管,重慶服務器租用等四川省內主機托管與主機租用業(yè)務;數(shù)據(jù)中心含:雙線機房,BGP機房,電信機房,移動機房,聯(lián)通機房。

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)  //這是典型的函數(shù)式編程

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

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

2、filter

過濾，選擇函數(shù)，

val filterRdd = mapRdd.filter(_ > 5)

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

3、flatmap加上reduceBykey

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

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

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

wordcount.collect //可以得到一個數(shù)組

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 unionrdd2 //結果是把兩個list合并成一個，List(('a',1),(‘a’, 2),('b',1),(‘b’, 2))

6、Join

笛卡爾積的干活，小組循環(huán)賽

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

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

val result_union = rdd1 joinrdd2 //結果是把兩個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的個數(shù)

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

rdd.count //計算行數(shù)

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

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

2、collect

collect函數(shù)可以提取出所有rdd里的數(shù)據(jù)項

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

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

val result=rdd1 union rdd2

使用collect操作查看一下執(zhí)行結果

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 的數(shù)據(jù)

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

另外有需要云服務器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內外云服務器15元起步，三天無理由+7*72小時售后在線，公司持有idc許可證，提供“云服務器、裸金屬服務器、高防服務器、香港服務器、美國服務器、虛擬主機、免備案服務器”等云主機租用服務以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務可用性高、性價比高”等特點與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應用場景需求。

網站名稱：Spark的transformation和action的操作學習筆記-創(chuàng)新互聯(lián)
網站路徑：http://weahome.cn/article/giddi.html