S-JIS[2014-09-02] 変更履歴

Spark Streaming

Apache SparkのStreamingのメモ。

概要

Spark Streamingは、流れてくるデータ（ストリーム）を処理する機能。

次々に流れてくるデータを（短い間隔で）繰り返しバッチ処理する。
大抵は結果をファイルシステム上に格納する。バッチ処理の都度書き込まれるので、ファイルが増えていくことになる。

短い間隔でバッチ処理を行う形なので、高スループット（単位時間当たりの処理能力が高い）であるが、レスポンス（応答時間）は遅くなる。

Spark Streamingを扱う場合、StreamingContextクラスを使う。
入力データはDStream（discretized stream、離散ストリーム）というクラスで扱う。
DStreamはRDDではないが、RDDと似たメソッドを持っている。（それらのメソッドを使って処理を記述することを「バッチ処理」と呼んでいるように思う）

import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._

  val sc: SparkContext = 〜
  val batchDuration = Seconds(1)
  val ssc = new StreamingContext(sc, batchDuration)

StreamingContextインスタンスはSparkContextを使って生成する。
引数のbatchDurationには、入力のストリームデータを後続のバッチ処理に送る間隔（スライド間隔）を指定する。

処理（メソッド）によっては、ウィンドウ間隔を指定するものもある。

短いバッチ処理を繰り返し実行していくことになるが、バッチ1回分の対象データをRDDで保持する形になっているようだ。

Spark Streamingの例

TCPソケットからテキストを読み込んでファイルに出力する例。

import org.apache.spark.SparkContext
import org.apache.spark.streaming._

object StreamingExample {

  val sc = new SparkContext("local[2]", "streaming-example")
  val ssc = new StreamingContext(sc, Seconds(1))

  def main(args: Array[String]): Unit = {
    // バッチ処理
    val stream1 = ssc.socketTextStream("localhost", 56789) // DStream[String]
    val stream2 = stream1.flatMap(_.split(" "))
    stream2.saveAsTextFiles("/tmp/spark/streaming1");

    // ストリーミング処理を3秒間実行
    ssc.start()
    try {
      Thread.sleep(3 * 1000)
    } finally {
      ssc.stop(true) // SparkContextもstopする
    }
  }
}

ストリームの読み込みを開始（ssc.start()）する前に、DStreamのアクション（処理内容）を記述しておく。
この例では「socketTextStreamでサーバーに接続してStringを受け取り、変換を行ってファイルに書き込む」という処理を記述している。
（RDDを記述するのとは異なり、DStreamの場合は、この時点ではまだ実際の処理は行われない。
 いわば、バッチ処理の内容を事前に定義（宣言）しているような感じ）

DStreamの処理内容を記述したら、StreamingContextのstart()でストリーム読み込みを開始する。
これによってDStreamの処理が実際に行われる。
stop()で読み込みを終了する。

サーバー側のサンプルは以下のような感じ。（何の変哲も無いTCPサーバー）

import java.net.ServerSocket

import scala.concurrent.ops.spawn

object StramingExampleServer {

  def main(args: Array[String]): Unit = {
    val server = new ServerSocket(56789)

    while (true) {
      val socket = server.accept()

      spawn {
        val os = socket.getOutputStream()
        try {
          for (i <- 0 to 9) {
            os.write(s"[${i}]\n".getBytes("UTF-8"))
            os.write(s"${new java.util.Date()}\n".getBytes("UTF-8"))

            Thread.sleep(200)
          }
        } finally {
          os.close()
        }
      }
    }
  }
}

StreamingContext

StreamingContextは、Spark Streaming用の基本的なクラス。

import org.apache.spark.streaming.StreamingContext

StreamingContextのコンストラクター

引数のbatchDurationには、入力のストリームデータを後続のバッチ処理に送る間隔を指定する。
→Durationクラス

StreamingContextには以下のようなメソッドがある。

• DStream生成系
• 実行系
• 設定系

DStream生成系

実行系

設定系

DStream

DStream（discretized stream、離散ストリーム）は、Spark Streaming用の“RDD相当”のクラス。
（直接RDDを継承してはいないが、似たメソッドを持っている）

DStream内部では、スライド間隔に応じたデータ群をRDDとして扱っているようだ。

変換処理

新しいDStreamを返す処理。

reduceやcountといった複数データを扱う（かつウィンドウ間隔を指定しない）メソッドの場合、処理対象データの範囲は、前回処理以降に来たデータになる模様。

アクション

ストリーム処理の結果を保存する処理。（RDDに合わせて、便宜上アクションという分類にしてみた）
Spark Streamingの場合はデータが次々と来るので、基本的にはファイルに出力することになるだろう。

（RDDのアクション系メソッドの場合、そのメソッドを実行したら実際に処理が行われるが、Spark Streamingの場合は、これらのメソッドを呼び出しても実際の処理は行われない。
StreamingContextをstartすると（定期的に）実行される）

printやforeachRDD・saveといった処理は、スライド間隔（slideDuration）の時間ごとに実行される。

永続化（キャッシュ）処理

その他

Duration

Durationは間隔[ms]を表すクラス。
Spark Streamingで使用するDurationはScala標準のscala.concurrent.durationではなく、Spark Streaming独自のorg.apache.spark.streaming.Durationである。

import org.apache.spark.streaming._

Durationを生成する為のオブジェクトとして以下のものが用意されている。

Durationには四則演算+,-,*,/や比較演算<,<=,>=,>等のメソッドが定義されている。

Time

Timeは日時を表すクラス。
（内部では値をミリ秒で保持している）

import org.apache.spark.streaming.Time

val time1 = Time(System.currentTimeMillis())
val time2 = time1 + Seconds(1)

println(new java.util.Date(time2.milliseconds))

Timeには加減算+,-や比較演算<,<=,>=,>等のメソッドが定義されている。