簡介

??Spark Streaming是Spark核心API的擴展，可以實現(xiàn)可伸縮、高吞吐量、具備容錯機制的實時流時數(shù)據(jù)的處理。支持多種數(shù)據(jù)源，比如Kafka、Flume、Twitter、ZeroMQ、Kinesis 以及TCP sockets。

??可以使用諸如map、reduce、join和window等高級函數(shù)進行復(fù)雜算法（比如，機器學(xué)習(xí)和圖計算）的處理。最后還可以將處理結(jié)果存儲到文件系統(tǒng)，數(shù)據(jù)庫和儀表盤。

spark streaming.png

架構(gòu)與抽象

抽象

??Spark Streaming接收實時流的數(shù)據(jù)，并根據(jù)一定的時間間隔拆分成一批批的數(shù)據(jù)，然后通過Spark Engine處理這些批數(shù)據(jù)，最終得到處理后的一批批結(jié)果數(shù)據(jù)。

streaming-flow.png

??Spark Streaming提供了一個叫做DStream(discretized stream，離散流)的抽象概念，DStream由一系列的RDD組成，表示每個批次中連續(xù)的數(shù)據(jù)流。DStream可以從輸入源（比如，Kafka、Flume、Kinesis等）中創(chuàng)建，也可以從其他DStream中使用高級算子操作轉(zhuǎn)換生成。

streaming-dstream.png

??DStream的所有操作其實都是對DStream中所有RDD的操作。比如，在單詞統(tǒng)計案例中，flatMap轉(zhuǎn)化操作會應(yīng)用到每個行RDD上來生成單詞RDD。

streaming-dstream-ops.png

架構(gòu)

架構(gòu).jpg

• Receiver：Spark Streaming內(nèi)置的數(shù)據(jù)流接收器或自定義接收器，用于從數(shù)據(jù)源接收源源不斷的數(shù)據(jù)流。

• CurrentBuffer：用于緩存輸入流接收器接收的數(shù)據(jù)流。

• BlockIntervalTimer：一個定時器，用于將CurrentBuffer中緩存的數(shù)據(jù)流封裝為Block后放入blocksForPushing隊列中。

• BlocksForPushing：待處理的Block

• BlockPushingThread：此線程每隔100毫秒從BlocksForPushing隊列中取出一個Block存入存儲系統(tǒng)，并緩存到ReceivedBlockQueue隊列中。

• Block Batch：Block批次，按照批次時間間隔，從ReceivedBlockQueue隊列中獲取一批Block。

• JobGenerator：Job生成器，用于給每一批Block生成一個Job。

DStream 轉(zhuǎn)化操作

??DStream轉(zhuǎn)化操作分為無狀態(tài)（stateless）和有狀態(tài)（stateful）兩種。

• 無狀態(tài)轉(zhuǎn)化操作中，每個批次的處理不依賴于之前批次的數(shù)據(jù)。

• 有狀態(tài)轉(zhuǎn)化操作需要使用之前批次的數(shù)據(jù)或中間結(jié)果來計算當(dāng)前批次的數(shù)據(jù)。

無狀態(tài)轉(zhuǎn)化操作

??無狀態(tài)轉(zhuǎn)化操作就是把簡單的RDD轉(zhuǎn)化操作應(yīng)用到每個批次上，轉(zhuǎn)化DStream中的每個RDD。

常用的無狀態(tài)轉(zhuǎn)化操作

• 使用map()和reduceByKey()在每個時間區(qū)間中對日志根據(jù)IP地址進行計數(shù)。

  • scala

  //假設(shè)ApacheAccessingLog是用來從Apache日志中解析條目的工具類
  val accessLogDStream = logData.map(line => ApacheAccessingLog.parseFromLogLine(line))
  val ipDStream = accessLogsDStream.map(entry => (entry.getIpAddress(), 1)
  val ipCountsDStream = ipDStream.reduceByKey((x,y) => x+y)
  

  • java

```
//假設(shè)ApacheAccessingLog是用來從Apache日志中解析條目的工具類
static final class IpTuple implements PairFunction<ApacheAccessLog, String, Long> {
    public Tuple2<String, Long> call(ApacheAccessLog log) {
        return new Tuple2<>(log.getIpAddress(), 1L);
    }
}

JavaDStream<ApacheAccessLog> accessLogDStream = logData.map(new ParseFromLogLine());
JavaPairDStream<String, Long> ipDStream = accessLogDStream.mapToPair(new IpTuple());
JavaPairDStream(String, Long) ipCountsDStream = ipDStream.reduceByKey(new LongSumReducer());
```

• 以IP地址為鍵，將請求計數(shù)的數(shù)據(jù)和傳輸數(shù)據(jù)量的數(shù)據(jù)連接起來

  • scala

  val ipBytesDStream = accessLogsDStream.map(entry => (entry.getIpAddress(), entry.getContentSize()))
  val ipBytesSumDStream = ipBytesDStream.reduceByKey((x,y) => x+y)
  val ipBytesRequestCountDStream = ipCountsDStream.join(ipBytesSumDStream)
  

  • java

```
JavaPairDStream<String, Long> ipBytesDStream = accessLogsDStream.mapToPair(new IpContentTuple());
JavaPairDStream<String, Long> ipBytesSumDStream = ipBytesDStream.reduceByKey(new LongSumReducer());
JavaPairDStream<String, Tuple2<Long,Long>> ipBytesRequestCountDStream = ipCountsDStream.join(ipBytesSumDStream);
```

• 使用transform()操作實現(xiàn)自定義轉(zhuǎn)化操作，從日志記錄中提取異常值。

  • scala

  val outlierDStream = accessLogsDStream.transform{
      rdd => extractOutliers(rdd)
  }
  

  • java

  JavaPairDStream<String, Long> ipRawDStream = accessLogsDStream.transform(
      new Function<JavaPairRDD<ApacheAccessLog>, JavaRDD<ApacheAccessLog>>() {
          public JavaPairRDD<ApacheAccessLog> call(JavaRDD<ApacheAccessLog> rdd) {
              return extractOutliers(rdd);
          }
      }
  );
  

有狀態(tài)轉(zhuǎn)化操作

??DStream的有狀態(tài)轉(zhuǎn)化操作是跨時間區(qū)間跟蹤數(shù)據(jù)的操作，先前批次的數(shù)據(jù)也被用來在新的批次中計算結(jié)果。

??有狀態(tài)轉(zhuǎn)化操作主要有兩種類型：滑動窗口和updateStateByKey()。前者以一個時間階段為滑動窗口進行操作，后者用來跟蹤每個鍵的狀態(tài)變化。

設(shè)置檢查點

??有狀態(tài)轉(zhuǎn)化操作需要在StreamingContext中打開檢查點機制確保容錯性。

ssc.checkpoint("hdfs://...")

基于窗口的轉(zhuǎn)化操作

簡介

??基于窗口的操作會在一個比StreamingContext批次間隔更長的時間范圍內(nèi)，通過整合多個批次的結(jié)果，計算出整個窗口的結(jié)果。

??基于窗口的轉(zhuǎn)化操作需要兩個參數(shù)，分別是窗口時長和滑動時長。兩者都是批次間隔的整數(shù)倍。

• 窗口時長：控制每次計算最近的windowDuration/batchInterval個批次的數(shù)據(jù)。

• 滑動步長：默認(rèn)值與批次間隔相等。用來控制對新DStream進行計算的時間間隔。

簡單案例

• 使用window()對窗口進行計數(shù)

  • scala

  val accessLogsWindow = accessLogsDStream.window(Seconds(30), Seconds(10))
  val windowCounts = accessLogsWindow.count()
  

  • java

```
JavaDStream<ApacheAccessLog> accessLogsWindow = accessLogsDStream.window(Durations.seconds(30), Duration.seconds(10));
JavaDStream<Integer> windowCounts = accessLogsWindow.count();
```

• 使用reduceByKeyAndWindow對每個IP地址的訪問量計數(shù)

  • scala

  val ipDStream = accessLogsDStream.map(logEntry => (logEntry.getIpAddress(), 1))
  val ipCountDStream = ipDStream.reduceByKeyAndWindow(
      {(x,y) => x+y}, //加入新進入窗口的批次中的元素
      {(x,y) => x-y}, //移除離開窗口的老批次中的元素
      Seconds(30), //窗口時長
      Seconds(10) //滑動步長
  )
  

  • java

```
class ExtractIp extends PairFunction<ApacheAccessLog, String, Long> {
    public Tuple2<String, Long> call(ApacheAccessLog entry) {
        return new Tuple2(entry.getIpAddress(), 1L);
    }
}

class AddLongs extends Function2<Long, Long, Long>() {
    public Long call(Long v1, Long v2) {
        return v1 + v2;
    }
}

class SubtractLongs extends Function2<Long, Long, Long>() {
    public Long call(Long v1, Long v2) {
        return v1 - v2;
    }
}

JavaPairDStream<String, Long> ipAddressPairDStream = accessLogsDStream.mapToPair(new ExtractIp());
JavaPairDStream<String, Long> ipCountDStream = ipAddressPairDStream.reduceByKeyAndWindow(
    new AddLongs(), //加上新進入窗口的批次中的元素
    new SubtractLongs(), //移除離開窗口的老批次中的元素
    Durations.seconds(30), //窗口時長
    Durations.seconds(10) //滑動步長
)
```

• 使用countByWindow和countByValueAndWindow對窗口計數(shù)

  • scala

  val ipDStream = accessLogsDStream.map{entry => entry.getIpAddress()}
  val ipAddressRequestCount = ipDStream.countByValueAndWindow(Seconds(30), Seconds(10))
  val requestCount = accessLogsDStream.countByWindow(Seconds(30), Seconds(10))
  

  • java

```
JavaDStream<String> ip = accessLogsDStream.map(new Function<ApacheAccessLog, String>() {
   public String call(ApacheAccessLog entry) {
        return entry.getIpAddress();
   }
});

JavaDStream<Long> requestCount = accessLogsDStream.countByWindow(Dirations.seconds(30), Durations.seconds(10));
JavaPairDStream<String, Long> ipAddressRequestCount = ip.countByValueAndWindow(Dirations.seconds(30), Durations.seconds(10));
```

updateStateByKey轉(zhuǎn)化操作

簡介

??updateStateByKey提供了跨批次維護狀態(tài)的功能，用于鍵值對形式的DStream。

??updateStateByKey提供了一個update(events, oldState)函數(shù)，接收與某鍵相關(guān)的事件及該鍵之前對應(yīng)的狀態(tài)，返回該鍵對應(yīng)的新狀態(tài)。

• events：當(dāng)前批次中收到的事件列表
• oldState：一個可選的狀態(tài)對象，存放在Option內(nèi)；如果一個鍵沒有之前的狀態(tài)，這個值為空。
• newState：由函數(shù)返回，也以O(shè)ption形式存在；可以返回一個空的Option表示刪除該狀態(tài)。

簡單案例

??使用updateStateByKey()跟蹤日志消息中各HTTP響應(yīng)代碼的計數(shù)。

• scala

def updateRunningSum(values: Seq[Long], state: Option[Long]) = {
    Some(state.getOrElse(0L) + values.size)
}

val responseCodeDStream = accessLogsDStream.map(log => (log.getResponseCode(), 1L))
val responseCodeCountDStream = responseCodeDStream.updateStateByKey(updateRunningSum _)

• java

class UpdateRunningSum implements Function2<List<Long>, Optional<Long>, Optional<Long>> {
    public Optional<Long> call(List<Long> nums, Optional<Long> current) {
        long sum = current.or(0L);
        return Optional.of(sum + nums.size());
    }
};

JavaPairDStream<Integer, Long> responseCodeCountDStream = accessLogsDStream.mapToPair(
    new PairFunction<ApacheAccessLog, Integer, Long>() {
        public Tuple2<Integer, Long> call(ApacheAccessLog log) {
            return new Tuple2(log.getResponseCode(), 1L);
        }
    }
).updateStateByKey(new UpdateRunningSum());

DStream 行動操作

??DStream行動操作同RDD的行動操作。比如，將DStream保存為SequenceFile文件。

• scala

val writableIpAddressRequestCount = ipAddressRequestCount.map{
    (ip, count) => <new Text(ip), new LongWritable(count))
}

writableIpAddressRequestCount.saveAsHadoopFiles[SequenceFileOutputFormat[Text, LongWritable]]("outputDir", "txt")
}

• java

JavaPairDStream<Text, LongWritable> writableDStream = ipDStream.mapToPair(
    new PairFunction<Tuple2<String, Long>, Text, LongWritable>() {
        public Tuple2<Text, LongWritable> call(Tuple2<String, Long> e) {
            return new Tuple2(new Text(e._1()), new LongWritable(e._2()));
        }
    }
);

writableDStream.saveAsHadoopFiles("outputDir", "txt", Text.class, LongWritable.class, SequenceFileOutputFormat.class);