Spark Streaming--実戦編

要約:
     Sprak StreamingはSaprk APの拡張に属し、リアルタイムデータストリーム(live data streams)の拡張性、高スループットフォールトトレランス(fault-tolerant)のストリーム処理をサポートする.KafKa,Flume,ZeroMQ Kinesisから TwitterやTCPソケットのデータソース、処理結果データはファイルシステムデータベースフィールドdashboardsなどに格納できます.
 
DStreamプログラミングモデル
DstreamはSpark streamingにおける高度な抽象連続データストリームであり、このデータソースは外部から(例えばKafKa Flumeなど)取得してもよいし、入力ストリームから取得してもよいし、他のDStream上で高度な操作を行うことによって作成してもよいし、DStreamは時間系列の連続したRDDによって表されるので、1つのDStreamは1つのRDDsのシーケンスと見なすことができる.
 
DStreamアクション
1.ソケットストリーム:Socketポートを傍受してデータを受信します.
入力ストリームとしてScalaライタによって一連の文字が生成されます.
GenerateChar:

object GenerateChar {
  def generateContext(index : Int) : String = {
    import scala.collection.mutable.ListBuffer
    val charList = ListBuffer[Char]()
    for(i <- 65 to 90)
      charList += i.toChar
    val charArray = charList.toArray
    charArray(index).toString
  }
  def index = {
    import  java.util.Random
    val rdm = new Random
    rdm.nextInt(7) 
  }
  def main(args: Array[String]) {
    val listener = new ServerSocket(9998)
    while(true){
      val socket = listener.accept()
      new Thread(){
        override def run() = {
          println("Got client connected from :"+ socket.getInetAddress)
          val out = new PrintWriter(socket.getOutputStream,true)
          while(true){
            Thread.sleep(500)
            val context = generateContext(index)  //                 
            println(context)
            out.write(context + '
')
            out.flush()
          }
          socket.close()
        }
      }.start()
    }
  }
}

ScoketStreaming:

object ScoketStreaming {
  def main(args: Array[String]) {
    //       StreamingContext， 2     
    val conf = new SparkConf().setMaster("local[2]").setAppName("ScoketStreaming")
    val sc = new StreamingContext(conf,Seconds(10))   //  10         
    //     DStream，  master:9998
    val lines = sc.socketTextStream("master",9998)
    val words = lines.flatMap(_.split(" "))
    val wordCounts = words.map(x => (x , 1)).reduceByKey(_ + _)
    wordCounts.print()
    sc.start()         //    
    sc.awaitTermination()   //        ，        
  }
}

実行結果:
GenerateCharが生成したデータは次のとおりです.

Got client connected from :/192.168.31.128
C
G
B
C
F
G
D
G
B

ScoketStreamingの実行結果:

-------------------------------------------
Time: 1459426750000 ms
-------------------------------------------
(B,1)
(G,1)
(C,1)
-------------------------------------------
Time: 1459426760000 ms
-------------------------------------------
(B,5)
(F,3)
(D,4)
(G,3)
(C,3)
(E,1)

注意:ローカルで実行されている場合は、setMasterのパラメータはlocal[n],n>1でなければなりません.公式サイトで説明します.

   When running a Spark Streaming program locally, do not use “local” or “local[1]” as the master URL. Either ofthese means that only one thread 
will be used for running tasks locally. If you are using a input DStream based on a receiver (e.g. sockets, Kafka, Flume, etc.), then the single 
thread will be used to run the receiver,leaving no thread for processing the received data.
 
         Spark Streaming   ,Master URL     "local" "local[1]",                       ，    Input DStream       
（ Kafka，Flume ），             ，                 。

クラスタ上で実行される場合、Spark streamingに割り当てられるコア数は、受信機よりも大きいデータでなければなりません.そうしないと、同じようにデータのみが受信され、処理できません.
 
2.ファイルフロー:Spark Streamingファイルシステムの変化を監視し、新しいファイルが追加されたら、それを読み込んでデータフローとする
次の点に注意してください.
  1.これらのファイルは同じ形式である
  2.これらのファイルは、原子移動またはファイル名の変更によりdataDirectoryで作成されます.
  3.これらのファイルを移動すると、これ以上変更することはできません.ファイルに内容を追加すると、これらの追加された新しいデータも読み込まれません.
FileStreaming:

object FileStreaming {
  def main(args: Array[String]) {
    val conf = new SparkConf().setMaster("local").setAppName("FileStreaming")
    val sc = new StreamingContext(conf,Seconds(5))
    val lines = sc.textFileStream("/home/hadoop/wordCount")
    val words = lines.flatMap(_.split(" "))
    val wordCounts = words.map(x => (x , 1)).reduceByKey(_ + _)
    sc.start()
    sc.awaitTermination()
  }
}

ファイルディレクトリにファイルを追加するとSparkは Streamingは自動的に読み込んで計算し、ローカルディレクトリHDFSや他のファイルシステムを読み取ることができます.
注:ファイル・ストリームは受信機を実行する必要がないため、コア数を割り当てる必要はありません.
 
3.RDDキューフロー:streamingContext.queueStream(queueOfRDD)を使用して、Spark StreamingアプリケーションをデバッグするためのRDDキューベースのDStreamを作成します.
QueueStream:プログラムは1秒ごとにRDDを作成し、Streamingは1秒ごとにデータを処理する

object QueueStream {
  def main(args: Array[String]) {
    val conf = new SparkConf().setMaster("local[2]").setAppName("queueStream")
    // 1        
    val ssc = new StreamingContext(conf,Seconds(1))
    //      push QueueInputDStream RDDs  
    val rddQueue = new mutable.SynchronizedQueue[RDD[Int]]()
    //    RDD         
    val inputStream = ssc.queueStream(rddQueue)
    val mappedStream = inputStream.map(x => (x % 10,1))
    val reduceStream = mappedStream.reduceByKey(_ + _)
    reduceStream.print
    ssc.start()
    for(i <- 1 to 30){
      rddQueue += ssc.sparkContext.makeRDD(1 to 100, 2)   //  RDD，       
      Thread.sleep(1000)                                  
    }
    ssc.stop()
  }
}

しゅつりょく

-------------------------------------------
Time: 1459595433000 ms // 1   
-------------------------------------------
(4,10)
(0,10)
(6,10)
(8,10)
(2,10)
(1,10)
(3,10)
(7,10)
(9,10)
(5,10)
............
............
-------------------------------------------
Time: 1459595463000 ms // 30   
-------------------------------------------
(4,10)
(0,10)
(6,10)
(8,10)
(2,10)
(1,10)
(3,10)
(7,10)
(9,10)
(5,10)

 
4.ステータス付き処理staefull
updateStateByKey操作:updateStateByKey操作を使用する場所はkeyの状態を保持し、継続的に更新できるようにするためである.この機能を使用するには、次の2つのステップがあります.
  1.任意のデータ型であってもよい状態を定義する
  2.ステータス更新関数を定義し、前のステータスに基づいてステータスを更新する方法を決定する関数を指定します.
 
同様にwordCountを例にとると,毎回の出力が積算される前のwordCountとは異なる.
StateFull:

object StateFull {
  def main(args: Array[String]) {
    //        
    val updateFunc = (values: Seq[Int], state: Option[Int]) => {
      val currentCount = values.foldLeft(0)(_ + _)
      val previousCount = state.getOrElse(0)
      Some(currentCount + previousCount)
    }
    val conf = new SparkConf().setMaster("local[2]").setAppName("stateFull")
    val sc = new StreamingContext(conf, Seconds(5))
    sc.checkpoint(".")    //     ，         ，         
    val lines = sc.socketTextStream("master", 9999)
    val words = lines.flatMap(_.split(" "))
    val wordDstream = words.map(x => (x, 1))
    val stateDstream = wordDstream.updateStateByKey[Int](updateFunc)
    stateDstream.print()
    sc.start()
    sc.awaitTermination()
  }
}

先にGenerateCharを実行してアルファベットを生成し、StateFullを実行します.結果は次のとおりです.

-------------------------------------------
Time: 1459597690000 ms
-------------------------------------------
(B,3)
(F,1)
(D,1)
(G,1)
(C,1)
-------------------------------------------
Time: 1459597700000 ms //       
-------------------------------------------
(B,5)
(F,3)
(D,4)
(G,4)
(A,2)
(E,5)
(C,4)

 
Spark Stramingの最大の利点は、データを処理するために粗粒度の処理方式(一度に小ロットのデータを処理する)を採用することであり、この特性も、RDD上の高度なDStreamを有するフォールトトレランス回復機構をより容易に実現することである.
抽象的であるため,RDDとの相互運用が極めて容易である.