本節(jié)主要內(nèi)容

本節(jié)部分內(nèi)容來(lái)自官方文檔：http://spark.apache.org/docs/latest/streaming-programming-guide.html

DStream Transformation操作

1. Transformation操作

TransformationMeaning

map(func)	對(duì)DStream中的各個(gè)元素進(jìn)行func函數(shù)操作，然后返回一個(gè)新的DStream.
flatMap(func)	與map方法類(lèi)似，只不過(guò)各個(gè)輸入項(xiàng)可以被輸出為零個(gè)或多個(gè)輸出項(xiàng)
filter(func)	過(guò)濾出所有函數(shù)func返回值為true的DStream元素并返回一個(gè)新的DStream
repartition(numPartitions)	增加或減少DStream中的分區(qū)數(shù)，從而改變DStream的并行度
union(otherStream)	將源DStream和輸入?yún)?shù)為otherDStream的元素合并，并返回一個(gè)新的DStream.
count()	通過(guò)對(duì)DStreaim中的各個(gè)RDD中的元素進(jìn)行計(jì)數(shù)，然后返回只有一個(gè)元素的RDD構(gòu)成的DStream
reduce(func)	對(duì)源DStream中的各個(gè)RDD中的元素利用func進(jìn)行聚合操作，然后返回只有一個(gè)元素的RDD構(gòu)成的新的DStream.
countByValue()	對(duì)于元素類(lèi)型為K的DStream，返回一個(gè)元素為（K,Long）鍵值對(duì)形式的新的DStream，Long對(duì)應(yīng)的值為源DStream中各個(gè)RDD的key出現(xiàn)的次數(shù)
reduceByKey(func, [numTasks])	利用func函數(shù)對(duì)源DStream中的key進(jìn)行聚合操作，然后返回新的（K，V）對(duì)構(gòu)成的DStream
join(otherStream, [numTasks])	輸入為（K,V)、（K,W）類(lèi)型的DStream，返回一個(gè)新的（K，（V，W）類(lèi)型的DStream
cogroup(otherStream, [numTasks])	輸入為（K,V)、（K,W）類(lèi)型的DStream，返回一個(gè)新的 (K, Seq[V], Seq[W]) 元組類(lèi)型的DStream
transform(func)	通過(guò)RDD-to-RDD函數(shù)作用于源碼DStream中的各個(gè)RDD，可以是任意的RDD操作，從而返回一個(gè)新的RDD
updateStateByKey(func)	根據(jù)于key的前置狀態(tài)和key的新值，對(duì)key進(jìn)行更新，返回一個(gè)新?tīng)顟B(tài)的DStream

具體示例：

//讀取本地文件~/streaming文件夾val lines = ssc.textFileStream(args(0))val words = lines.flatMap(_.split(" "))val wordMap = words.map(x => (x, 1))val wordCounts=wordMap.reduceByKey(_ + _)val filteredWordCounts=wordCounts.filter(_._2>1)val numOfCount=filteredWordCounts.count()val countByValue=words.countByValue()val union=words.union(word1)val transform=words.transform(x=>x.map(x=>(x,1))) //顯式原文件lines.print() //打印flatMap結(jié)果words.print() //打印map結(jié)果wordMap.print() //打印reduceByKey結(jié)果wordCounts.print() //打印filter結(jié)果filteredWordCounts.print() //打印count結(jié)果numOfCount.print() //打印countByValue結(jié)果countByValue.print() //打印union結(jié)果union.print() //打印transform結(jié)果transform.print()

• 1

下面的代碼是運(yùn)行時(shí)添加的文件內(nèi)容

root@sparkmaster:~/streaming# echo "A B C D" >> test12.txt; echo "A B" >> test12.txt

• 1

下面是前面各個(gè)函數(shù)的結(jié)果

------------------------------------------- lines.print() ------------------------------------------- A B C D A B------------------------------------------- flatMap結(jié)果 ------------------------------------------- A B C D A B------------------------------------------- map結(jié)果 ------------------------------------------- (A,1) (B,1) (C,1) (D,1) (A,1) (B,1)------------------------------------------- reduceByKey結(jié)果 ------------------------------------------- (B,2) (D,1) (A,2) (C,1)------------------------------------------- filter結(jié)果 ------------------------------------------- (B,2) (A,2)------------------------------------------- count結(jié)果 ------------------------------------------- 2------------------------------------------- countByValue結(jié)果 ------------------------------------------- (B,2) (D,1) (A,2) (C,1)------------------------------------------- union結(jié)果 ------------------------------------------- A B C D A B A B C D ...------------------------------------------- transform結(jié)果 ------------------------------------------- (A,1) (B,1) (C,1) (D,1) (A,1) (B,1)

• 1

示例2：
上節(jié)課中演示的WordCount代碼并沒(méi)有只是對(duì)輸入的單詞進(jìn)行分開(kāi)計(jì)數(shù)，沒(méi)有記錄前一次計(jì)數(shù)的狀態(tài)，如果想要連續(xù)地進(jìn)行計(jì)數(shù)，則可以使用updateStateByKey方法來(lái)進(jìn)行。下面的代碼主要給大家演示如何updateStateByKey的方法。

import org.apache.spark.SparkConf import org.apache.spark.HashPartitioner import org.apache.spark.streaming._object StatefulNetworkWordCount {def main(args: Array[String]) {if (args.length < 2) {System.err.println("Usage: StatefulNetworkWordCount <hostname> <port>")System.exit(1)}//函數(shù)字面量，輸入的當(dāng)前值與前一次的狀態(tài)結(jié)果進(jìn)行累加val updateFunc = (values: Seq[Int], state: Option[Int]) => {val currentCount = values.sumval previousCount = state.getOrElse(0)Some(currentCount + previousCount)}//輸入類(lèi)型為K,V,S,返回值類(lèi)型為K,S//V對(duì)應(yīng)為帶求和的值，S為前一次的狀態(tài)val newUpdateFunc = (iterator: Iterator[(String, Seq[Int], Option[Int])]) => {iterator.flatMap(t => updateFunc(t._2, t._3).map(s => (t._1, s)))}val sparkConf = new SparkConf().setAppName("StatefulNetworkWordCount").setMaster("local[4]")//每一秒處理一次val ssc = new StreamingContext(sparkConf, Seconds(1))//當(dāng)前目錄為checkpoint結(jié)果目錄，后面會(huì)講checkpoint在Spark Streaming中的應(yīng)用ssc.checkpoint(".")//RDD的初始化結(jié)果val initialRDD = ssc.sparkContext.parallelize(List(("hello", 1), ("world", 1)))//使用Socket作為輸入源，本例ip為localhost，端口為9999val lines = ssc.socketTextStream(args(0), args(1).toInt)//flatMap操作val words = lines.flatMap(_.split(" "))//map操作val wordDstream = words.map(x => (x, 1))//updateStateByKey函數(shù)使用val stateDstream = wordDstream.updateStateByKey[Int](newUpdateFunc,new HashPartitioner (ssc.sparkContext.defaultParallelism), true, initialRDD)stateDstream.print()ssc.start()ssc.awaitTermination()} }

下圖是初始時(shí)的值：

使用下列命令啟動(dòng)netcat server

root@sparkmaster:~/streaming# nc -lk 9999

然后輸入

root@sparkmaster:~/streaming# nc -lk 9999 hello

將得到下圖的結(jié)果

然后再輸入world，

root@sparkmaster:~/streaming# nc -lk 9999 hello world

• 1

則將得到下列結(jié)果

總結(jié)

以上是生活随笔為你收集整理的Spark Streaming 实战案例（二) Transformation操作的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。