本篇文章給大家分享的是有關(guān)DStreams上的輸出操作是怎樣的，小編覺得挺實用的，因此分享給大家學(xué)習(xí)，希望大家閱讀完這篇文章后可以有所收獲，話不多說，跟著小編一起來看看吧。

成都創(chuàng)新互聯(lián)長期為上1000家客戶提供的網(wǎng)站建設(shè)服務(wù)，團隊從業(yè)經(jīng)驗10年，關(guān)注不同地域、不同群體，并針對不同對象提供差異化的產(chǎn)品和服務(wù)；打造開放共贏平臺，與合作伙伴共同營造健康的互聯(lián)網(wǎng)生態(tài)環(huán)境。為黑山企業(yè)提供專業(yè)的網(wǎng)站設(shè)計、成都網(wǎng)站建設(shè)，黑山網(wǎng)站改版等技術(shù)服務(wù)。擁有十年豐富建站經(jīng)驗和眾多成功案例,為您定制開發(fā)。

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

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

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

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

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

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

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

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

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

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

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
      })
  })

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

其它需要注意的地方：

• 輸出操作通過懶執(zhí)行的方式操作DStreams，正如RDD action通過懶執(zhí)行的方式操作RDD。具體地看，RDD actions和DStreams輸出操作接收數(shù)據(jù)的處理。因此，如果你的應(yīng)用程序沒有任何輸出操作或者 用于輸出操作dstream.foreachRDD()，但是沒有任何RDD action操作在dstream.foreachRDD()里面，那么什么也不會執(zhí)行。系統(tǒng)僅僅會接收輸入，然后丟棄它們。

• 默認(rèn)情況下，DStreams輸出操作是分時執(zhí)行的，它們按照應(yīng)用程序的定義順序按序執(zhí)行。

以上就是DStreams上的輸出操作是怎樣的，小編相信有部分知識點可能是我們?nèi)粘９ぷ鲿姷交蛴玫降?。希望你能通過這篇文章學(xué)到更多知識。更多詳情敬請關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道。