13.spark streaming之快速入門

簡介

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

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

架構與抽象

抽象

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

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

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

架構

• 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é)果來計算當前批次的數(shù)據(jù)。

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

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

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

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

  • scala

  //假設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

  //假設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)變化。

設置檢查點

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

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

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

簡介

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

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

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

• 滑動步長：默認值與批次間隔相等。用來控制對新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 ipAddre***equestCount = 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> ipAddre***equestCount = ip.countByValueAndWindow(Dirations.seconds(30), Durations.seconds(10));

updateStateByKey轉(zhuǎn)化操作

簡介

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

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

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

簡單案例

??使用updateStateByKey()跟蹤日志消息中各HTTP響應代碼的計數(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 writableIpAddre***equestCount = ipAddre***equestCount.map{
    (ip, count) => <new Text(ip), new LongWritable(count))
}

writableIpAddre***equestCount.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);

忠于技術，熱愛分享。歡迎關注公眾號：java大數(shù)據(jù)編程，了解更多技術內(nèi)容。