DStream的輸出操作

DStreams上的輸出操作

輸出操作允許DStream的操作推到如數(shù)據(jù)庫(kù)、文件系統(tǒng)等外部系統(tǒng)中。因?yàn)檩敵霾僮鲗?shí)際上是允許外部系統(tǒng)消費(fèi)轉(zhuǎn)換后的數(shù)據(jù)，它們觸發(fā)的實(shí)際操作是DStream轉(zhuǎn)換。目前，定義了下面幾種輸出操作：

利用foreachRDD的設(shè)計(jì)模式

dstream.foreachRDD是一個(gè)強(qiáng)大的原語(yǔ)，發(fā)送數(shù)據(jù)到外部系統(tǒng)中。然而，明白怎樣正確地、有效地用這個(gè)原語(yǔ)是非常重要的。下面幾點(diǎn)介紹了如何避免一般錯(cuò)誤。

• 經(jīng)常寫數(shù)據(jù)到外部系統(tǒng)需要建一個(gè)連接對(duì)象（例如到遠(yuǎn)程服務(wù)器的TCP連接），用它發(fā)送數(shù)據(jù)到遠(yuǎn)程系統(tǒng)。為了達(dá)到這個(gè)目的，開(kāi)發(fā)人員可能不經(jīng)意的在Spark驅(qū)動(dòng)中創(chuàng)建一個(gè)連接對(duì)象，但是在Spark worker中嘗試調(diào)用這個(gè)連接對(duì)象保存記錄到RDD中，如下：

  dstream.foreachRDD(rdd => {
      val connection = createNewConnection()  // executed at the driver
      rdd.foreach(record => {
          connection.send(record) // executed at the worker
      })
  })

這是不正確的，因?yàn)檫@需要先序列化連接對(duì)象，然后將它從driver發(fā)送到worker中。這樣的連接對(duì)象在機(jī)器之間不能傳送。它可能表現(xiàn)為序列化錯(cuò)誤（連接對(duì)象不可序列化）或者初始化錯(cuò)誤（連接對(duì)象應(yīng)該在worker中初始化）等等。正確的解決辦法是在worker中創(chuàng)建連接對(duì)象。

• 然而，這會(huì)造成另外一個(gè)常見(jiàn)的錯(cuò)誤-為每一個(gè)記錄創(chuàng)建了一個(gè)連接對(duì)象。例如：

  dstream.foreachRDD(rdd => {
      rdd.foreach(record => {
          val connection = createNewConnection()
          connection.send(record)
          connection.close()
      })
  })

通常，創(chuàng)建一個(gè)連接對(duì)象有資源和時(shí)間的開(kāi)支。因此，為每個(gè)記錄創(chuàng)建和銷毀連接對(duì)象會(huì)導(dǎo)致非常高的開(kāi)支，明顯的減少系統(tǒng)的整體吞吐量。一個(gè)更好的解決辦法是利用rdd.foreachPartition方法。為RDD的partition創(chuàng)建一個(gè)連接對(duì)象，用這個(gè)兩件對(duì)象發(fā)送partition中的所有記錄。

 dstream.foreachRDD(rdd => {
      rdd.foreachPartition(partitionOfRecords => {
          val connection = createNewConnection()
          partitionOfRecords.foreach(record => connection.send(record))
          connection.close()
      })
  })

這就將連接對(duì)象的創(chuàng)建開(kāi)銷分?jǐn)偟搅藀artition的所有記錄上了。

• 最后，可以通過(guò)在多個(gè)RDD或者批數(shù)據(jù)間重用連接對(duì)象做更進(jìn)一步的優(yōu)化。開(kāi)發(fā)者可以保有一個(gè)靜態(tài)的連接對(duì)象池，重復(fù)使用池中的對(duì)象將多批次的RDD推送到外部系統(tǒng)，以進(jìn)一步節(jié)省開(kāi)支。

  dstream.foreachRDD(rdd => {
      rdd.foreachPartition(partitionOfRecords => {
          // ConnectionPool is a static, lazily initialized pool of connections
          val connection = ConnectionPool.getConnection()
          partitionOfRecords.foreach(record => connection.send(record))
          ConnectionPool.returnConnection(connection)  // return to the pool for future reuse
      })
  })

需要注意的是，池中的連接對(duì)象應(yīng)該根據(jù)需要延遲創(chuàng)建，并且在空閑一段時(shí)間后自動(dòng)超時(shí)。這樣就獲取了最有效的方式發(fā)生數(shù)據(jù)到外部系統(tǒng)。

其它需要注意的地方：

• 輸出操作通過(guò)懶執(zhí)行的方式操作DStreams，正如RDD action通過(guò)懶執(zhí)行的方式操作RDD。具體地看，RDD actions和DStreams輸出操作接收數(shù)據(jù)的處理。因此，如果你的應(yīng)用程序沒(méi)有任何輸出操作或者用于輸出操作dstream.foreachRDD()，但是沒(méi)有任何RDD action操作在dstream.foreachRDD()里面，那么什么也不會(huì)執(zhí)行。系統(tǒng)僅僅會(huì)接收輸入，然后丟棄它們。
• 默認(rèn)情況下，DStreams輸出操作是分時(shí)執(zhí)行的，它們按照應(yīng)用程序的定義順序按序執(zhí)行。