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

讓客戶滿意是我們工作的目標(biāo)，不斷超越客戶的期望值來(lái)自于我們對(duì)這個(gè)行業(yè)的熱愛(ài)。我們立志把好的技術(shù)通過(guò)有效、簡(jiǎn)單的方式提供給客戶，將通過(guò)不懈努力成為客戶在信息化領(lǐng)域值得信任、有價(jià)值的長(zhǎng)期合作伙伴，公司提供的服務(wù)項(xiàng)目有：空間域名、網(wǎng)站空間、營(yíng)銷軟件、網(wǎng)站建設(shè)、額爾古納網(wǎng)站維護(hù)、網(wǎng)站推廣。

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

二、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才開(kāi)始執(zhí)行，是action，返回一個(gè)Array    Array(2,4,6,8,10,12)

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

2、filter

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

val filterRdd = mapRdd.filter(_ > 5)

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

3、flatmap加上reduceBykey

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

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

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

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

4、groupByKey

對(duì)文件按照空格進(jìn)行分割后，按照單詞進(jìn)行g(shù)roupByKey分組

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

使用collect查看一下結(jié)果

wordcount.collect

5、Union

2個(gè)合并成1個(gè)

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

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

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

6、Join

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

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

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

val result_union = rdd1 join rdd2 //結(jié)果是把兩個(gè)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))

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

三、action有哪些

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

1、count

計(jì)算rdd的個(gè)數(shù)

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

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

rdd.cache   //可以把rdd保留在內(nèi)存里面

rdd.count //計(jì)算行數(shù)，但是因?yàn)樯厦孢M(jìn)行了cache，這里速度會(huì)很快

2、collect

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

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í)行結(jié)果

3、reduce

map、reduce是hadoop的2個(gè)核心，map是映射，reduce是精簡(jiǎn)

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

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

4、lookup

查找的干活

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

rdd.lookup("a") //返回一個(gè)seq， (1, 2) 是把a(bǔ)對(duì)應(yīng)的所有元素的value提出來(lái)組成一個(gè)seq

5、save

查詢搜索結(jié)果排名第 1 點(diǎn)擊次序排在第 2 的數(shù)據(jù)

val rdd1 = sc.textFile("hdfs://192.168.0.10:9000/input/SogouQ2.txt").map(_.split("\t"))   //長(zhǎng)度為6錯(cuò)誤，好像日志不標(biāo)準(zhǔn)，有的為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/")