DStream中的轉(zhuǎn)換

DStream中的轉(zhuǎn)換（transformation）

和RDD類似，transformation允許從輸入DStream來的數(shù)據(jù)被修改。DStreams支持很多在RDD中可用的transformation算子。一些常用的算子如下所示：

最后兩個transformation算子需要重點介紹一下：

UpdateStateByKey操作

updateStateByKey操作允許不斷用新信息更新它的同時保持任意狀態(tài)。你需要通過兩步來使用它

• 定義狀態(tài)-狀態(tài)可以是任何的數(shù)據(jù)類型
• 定義狀態(tài)更新函數(shù)-怎樣利用更新前的狀態(tài)和從輸入流里面獲取的新值更新狀態(tài)

讓我們舉個例子說明。在例子中，你想保持一個文本數(shù)據(jù)流中每個單詞的運行次數(shù)，運行次數(shù)用一個state表示，它的類型是整數(shù)

def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
    val newCount = ...  // add the new values with the previous running count to get the new count
    Some(newCount)
}

這個函數(shù)被用到了DStream包含的單詞上

import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
// Create a local StreamingContext with two working thread and batch interval of 1 second
val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
val ssc = new StreamingContext(conf, Seconds(1))
// Create a DStream that will connect to hostname:port, like localhost:9999
val lines = ssc.socketTextStream("localhost", 9999)
// Split each line into words
val words = lines.flatMap(_.split(" "))
// Count each word in each batch
val pairs = words.map(word => (word, 1))
val runningCounts = pairs.updateStateByKey[Int](updateFunction _)

更新函數(shù)將會被每個單詞調(diào)用，newValues擁有一系列的1（從 (詞, 1)對而來），runningCount擁有之前的次數(shù)。要看完整的代碼，見例子

Transform操作

transform操作（以及它的變化形式如transformWith）允許在DStream運行任何RDD-to-RDD函數(shù)。它能夠被用來應(yīng)用任何沒在DStream API中提供的RDD操作（It can be used to apply any RDD operation that is not exposed in the DStream API）。例如，連接數(shù)據(jù)流中的每個批（batch）和另外一個數(shù)據(jù)集的功能并沒有在DStream API中提供，然而你可以簡單的利用transform方法做到。如果你想通過連接帶有預(yù)先計算的垃圾郵件信息的輸入數(shù)據(jù)流來清理實時數(shù)據(jù)，然后過了它們，你可以按如下方法來做：

val spamInfoRDD = ssc.sparkContext.newAPIHadoopRDD(...) // RDD containing spam information

val cleanedDStream = wordCounts.transform(rdd => {
  rdd.join(spamInfoRDD).filter(...) // join data stream with spam information to do data cleaning
  ...
})

事實上，你也可以在transform方法中用機器學(xué)習(xí)和圖計算算法

窗口(window)操作

Spark Streaming也支持窗口計算，它允許你在一個滑動窗口數(shù)據(jù)上應(yīng)用transformation算子。下圖闡明了這個滑動窗口。

如上圖顯示，窗口在源DStream上滑動，合并和操作落入窗內(nèi)的源RDDs，產(chǎn)生窗口化的DStream的RDDs。在這個具體的例子中，程序在三個時間單元的數(shù)據(jù)上進行窗口操作，并且每兩個時間單元滑動一次。這說明，任何一個窗口操作都需要指定兩個參數(shù)：

• 窗口長度：窗口的持續(xù)時間
• 滑動的時間間隔：窗口操作執(zhí)行的時間間隔

這兩個參數(shù)必須是源DStream的批時間間隔的倍數(shù)。

下面舉例說明窗口操作。例如，你想擴展前面的例子用來計算過去30秒的詞頻，間隔時間是10秒。為了達到這個目的，我們必須在過去30秒的pairs DStream上應(yīng)用reduceByKey操作。用方法reduceByKeyAndWindow實現(xiàn)。

// Reduce last 30 seconds of data, every 10 seconds
val windowedWordCounts = pairs.reduceByKeyAndWindow((a:Int,b:Int) => (a + b), Seconds(30), Seconds(10))

一些常用的窗口操作如下所示，這些操作都需要用到上文提到的兩個參數(shù)：窗口長度和滑動的時間間隔