流資料的特點

與一般的檔案（即內容已經固定）型資料來源相比，所謂的流資料擁有如下的特點

1.   資料一直處在變化中

2.   資料無法回退

3.   資料一直源源不斷的湧進

DStream

如果要用一句話來概括SparkStreaming的處理思路的話，那就是"將連續的資料持久化，離散化，然後進行批量處理"。

讓我們來仔細分析一下這麼作的原因。

·        資料持久化將從網路上接收到的資料先暫時儲存下來，為事件處理出錯時的事件重演提供可能，

·        離散化資料來源源不斷的湧進，永遠沒有一個盡頭，就像周星馳的喜劇中所說“崇拜之情如黃河之水綿綿不絕，一發而不可收拾”。既然不能窮盡，那麼就將其按時間分片。比如採用一分鐘為時間間隔，那麼在連續的一分鐘內收集到的資料集中儲存在一起。

·        批量處理將持久化下來的資料分批進行處理，處理機制套用之前的RDD模式

DStream可以說是對RDD的又一層封裝。如果開啟DStream.scala和RDD.scala，可以發現幾乎RDD上的所有operation在DStream中都有相應的定義。

作用於DStream上的operation分成兩類

1.   Transformation

2.   Output 表示將輸出結果，目前支援的有print,saveAsObjectFiles, saveAsTextFiles, saveAsHadoopFiles

DStreamGraph

有輸入就要有輸出，如果沒有輸出，則前面所做的所有動作全部沒有意義，那麼如何將這些輸入和輸出繫結起來呢？這個問題的解決就依賴於

  privateval inputStreams = new ArrayBuffer[InputDStream[_]]()
  privateval outputStreams = new ArrayBuffer[DStream[_]]()
  var rememberDuration: Duration = null
  var checkpointInProgress = false

outputStreams中的元素是在有Output型別的Operation作用於DStream上時自動新增到

outputStream區別於inputStream一個重要的地方就是會過載generateJob.

初始化流程

StreamingContext

StreamingContext是Spark Streaming初始化的入口點，主要的功能是根據入參來生成JobScheduler

設定InputStream

如果流資料來源來自於socket，則使用socketStream。如果資料來源來自於不斷變化著的檔案，則可使用fileStream

提交執行

StreamingContext.start()

資料處理

以socketStream為例，資料來自於socket。

SocketInputDstream啟動一個執行緒，該執行緒使用receive函式來接收資料

def receive() {                                                                                                          
    var socket: Socket = null                                                                                              
    try {                                                                                                                  
      logInfo("Connecting to " + host + ":" + port)                                                                        
      socket = new Socket(host, port)                                                                                      
      logInfo("Connected to " + host + ":" + port)                                                                         
      val iterator = bytesToObjects(socket.getInputStream())                                                               
      while(!isStopped && iterator.hasNext) {                                                                              
        store(iterator.next)                                                                                               
      }                                                                                                                    
      logInfo("Stopped receiving")                                                                                         
      restart("Retrying connecting to " + host + ":" + port)                                                               
    } catch {                                                                                                              
      case e: java.net.ConnectException =>                                                                                 
        restart("Error connecting to " + host + ":" + port, e)                                                             
      case t: Throwable =>                                                                                                 
        restart("Error receiving data", t)                                                                                 
    } finally {       
 
   if (socket != null) {                                                                                                
        socket.close()                                                                                                     
        logInfo("Closed socket to " + host + ":" + port)                                                                   
      }                                                                                                                    
    }                                                                                                                      
  }                                                                                                                        
}        

接收到的資料會被先儲存起來，儲存最終會呼叫到BlockManager.scala中的函式，那麼BlockManager是如何被傳遞到StreamingContext的呢？利用SparkEnv傳入的，注意StreamingContext建構函式的入參。

處理定時器

資料的儲存有是被socket觸發的。那麼已經儲存的資料被真正的處理又是被什麼觸發的呢？

記得在初始化StreamingContext的時候，我們指定了一個時間引數，那麼用這個引數會構造相應的重複定時器，一旦定時器超時，呼叫generateJobs函式。

privateval timer = new RecurringTimer(clock, ssc.graph.batchDuration.milliseconds, longTime => eventActor ! GenerateJobs(new Time(longTime)), "JobGenerator")

事件處理函式

 /** Processes all events */                                                                                              
  privatedef processEvent(event: JobGeneratorEvent) {                                                                     
    logDebug("Got event " + event)                                                                                         
    event match {                                                                                                          
      case GenerateJobs(time) => generateJobs(time)                                                                        
      case ClearMetadata(time) => clearMetadata(time)                                                                      
      case DoCheckpoint(time) => doCheckpoint(time)                                                                        
      case ClearCheckpointData(time) => clearCheckpointData(time)                                                          
    }                                                                                                                      
  }     

generteJobs

 privatedef generateJobs(time: Time) {                                                                                   
    SparkEnv.set(ssc.env)                                                                                                  
    Try(graph.generateJobs(time)) match {                                                                                  
      case Success(jobs) =>                                                                                                
        val receivedBlockInfo = graph.getReceiverInputStreams.map { stream =>                                              
          val streamId = stream.id                                                                                         
          val receivedBlockInfo = stream.getReceivedBlockInfo(time)                                                        
          (streamId, receivedBlockInfo)                                                                                    
        }.toMap                                                                                                            
        jobScheduler.submitJobSet(JobSet(time, jobs, receivedBlockInfo))                                                   
      case Failure(e) =>                                                                                                   
        jobScheduler.reportError("Error generating jobs for time " + time, e)                                              
    }                                                                                                                      
    eventActor ! DoCheckpoint(time)                                                                                        
  }          

 generateJobs->generateJob一路下去會呼叫到Job.run,在job.run中呼叫sc.runJob，在具體呼叫路徑就不一一列出。

 privateclassJobHandler(job: Job)extendsRunnable {
    def run() {
      eventActor ! JobStarted(job)
      job.run()
      eventActor ! JobCompleted(job)
    }
  }

DStream.generateJob函式中定義了jobFunc，也就是在job.run()中使用到的jobFunc

  private[streaming] def generateJob(time: Time): Option[Job] = {
    getOrCompute(time) match {
      case Some(rdd) => {
        val jobFunc = () => {
          val emptyFunc = { (iterator: Iterator[T]) => {} }
          context.sparkContext.runJob(rdd, emptyFunc)
        }
        Some(new Job(time, jobFunc))
      }
      case None => None
    }
  }

在這個流程中，DStreamGraph起到非常關鍵的作用，非常類似於TridentStorm中的graph.

在generateJob過程中，DStream會通過呼叫compute函式生成相應的RDD，SparkContext則是將基於RDD的抽象轉換成為多個stage來執行。在StreamingContext中一個重要的轉換是DStream到RDD的轉換，另一個重要的轉換是RDD到Stage及Task的轉換。