用Apache Spark進行大數據處理——第三部分：Spark流

介紹

流數據分析

• Apache Samza
• Storm
• Spark流

在本文中我們將專注于Spark流。

Spark流

Spark流是核心Spark API的擴展。Spark流使得基于實時數據流構建容錯性處理變得更加簡單。下面的圖1展示了Spark流是如何融入到整個Apache Spark生態系統中。

(點擊放大圖像)

圖1.具有Spark流庫的Spark生態系統

• Kafka
• Flume
• 推ter
• ZeroMQ
• Amazon's Kinesis
• TCP sockets

(點擊放大圖像)

圖2.Spark流如何工作

Spark流用例

• Uber ，車駕共享服務背后的公司，在他們的持續流式ETL管道中使用了Spark流以每天從其移動用戶處收集TB級的事件數據來進行實時遙測分析。
• Pinterest ，可視化書簽工具背后的公司，使用Spark流、MemSQL與Apache Kafka技術以實時地深入了解他們全球的用戶是怎樣使用Pins的。
• Netflix 使用Kafka與Spark流來構建一個實時在線電影推薦與數據監控解決 方案 ，該方案每天要處理來自于不同數據源的數十億條事件。

Spark流其他現實世界的樣例還包括：

• 供應鏈分析
• 實時安全情報操作以尋找威脅
• 廣告競價平臺
• 實時視頻分析，以幫助觀看者實現個性化與互動體驗

讓我們看一下Spark流的架構與API方法。若要編寫Spark流程序，我們需要知曉兩個組件：DStream與流上下文。

DStream

• map
• flatMap
• filter
• count
• reduce
• countByValue
• reduceByKey
• join
• updateStateByKey

流上下文

Spark流API

Spark流附帶了若干個用于處理數據流的API方法。有類似于RDD的操作，比如map、flatMap、filter、count、reduce、groupByKey、reduceByKey、sortByKey和join。它也提供了其他基于window與stateful操作的處理流數據的API。包括window、countByWindow、reduceByWindow、countByValueAndWindow、reduceByKeyAndWindow和updateStateByKey。Spark流庫當前支持Scala、Java和Python編程語言。這里是每個語言對應的Spark流API鏈接：

• Spark流Scala API
• Java API
• Python API

Spark編程的步驟

在我們討論樣例應用之前，先來看看Spark流編程中與眾不同的步驟：

• Spark流上下文被用于處理實時數據流。因此，第一步就是用兩個參數初始化流上下文對象，Spark上下文和切片間隔時間。切片間隔設置了流中我們處理輸入數據的更新窗口。一旦上下文被初始化，就無法再向已經存在的上下文中定義或添加新的計算。并且，在同一時間只有一個流上下文對象可以被激活。
• 當Spark流上下文被定義后，我們通過創建輸入DStreams來指定輸入數據源。在我們的樣例應用中，輸入數據源是一個使用了Apache Kafka分布式數據庫和消息系統的日志消息生成器。日志生成器程序創建隨機日志消息以模擬網絡服務器的運行時環境，作為各種網絡應用服務用戶而產生的流量，日志消息被持續不斷地生成。
• 使用map和reduce這樣的Spark流變換API為DStreams定義計算。
• 當流計算邏輯被定義好后，我們可以使用先前創建的流上下文對象中的start方法來開始接收并處理數據。
• 最終，我們使用流上下文對象的awaitTermination方法等待流數據處理完畢并停止它。

樣例應用

用例

樣例應用的用例是一個網絡服務器日志分析與統計的生成器。在樣例應用中，我們分析網絡服務器日志以計算如下統計信息，這些信息有助于進一步的數據分析和報表及面板的創建：

• 不同HTTP響應代碼的響應計數
• 響應內容大小
• 導致最高網絡流量的訪問客戶端的IP地址
• 最熱門的終端URL以識別那些比其他服務被訪問的更多服務

與本系列的前兩篇文章不同，在本文中我們將使用Java而非Scala來創建Spark程序。我們按照獨立應用的方式運行程序，而不是在控制臺窗口中運行代碼。在測試與產品環境中部署Spark程序也如此。Shell控制臺接口(使用Scala、Python或R語言)僅僅是用于開發者本地測試而已。

技術

在樣例程序中我們將使用如下的技術來演示如何使用Spark流庫處理實時數據流。

Zookeeper

Zookeeper 是一個為分布式應用提供可靠分布式協調的集中化的服務。Kafka，我們在樣例應用中使用的消息系統，依賴于Zoopkeeper在整個集群中的詳細設置。

Kafka

Spark流

我們將會使用Spark流Java API來接收數據流，計算日志統計信息并且運行查詢以回答諸如“最多網絡請求來自于哪個IP地址”這樣的問題。下面的表1展示了樣例應用中所使用的技術與工具以及他們的版本。

表1.Spark流樣例應用技術及工具

在圖3中演示了Spark流樣例應用中不同架構組件。

(點擊放大圖像)

圖3.Spark流樣例應用架構

Spark流應用運行時

為了在本地設置Java項目，可以從Github上下載 Databricks參考應用代碼 。一旦獲取了參考應用代碼，就需要兩個額外的Java類來運行我們的樣例應用。

• 日志生成器(SparkStreamingKafkaLogGenerator.java)
• 日志分析器(SparkStreamingKafkaLogAnalyzer.java)

Zookeeper命令：

在樣例程序中我使用的Zookeeper版本是3.4.6。為了啟動服務器，需要設置兩個環境變量，JAVA_HOME與ZOOKEEPER_HOME來指定JDK和Zookeeper各自的安裝目錄。然后導航到Zookeeper的home目錄并運行如下命令來啟動Zookeeper服務器。

bin\zkServer.cmd

如果你使用的是Linux環境，命令就是：

bin/zkServer.sh start

Kafka服務器命令：

在程序中使用的Kafka版本是2.10-0.9.00，基于Scala2.10版本。在Kafka中所使用的Scala版本是非常重要的，因為若是沒有使用恰當的版本的話，當執行Spark流程序時就會遇到運行時錯誤。這里是啟動Kafka服務器實例的步驟：

• 打開一個新的命令行窗口
• 設置JAVA_HONE與KAFKA_HOME環境變量
• 導航到Kafka的home目錄
• 運行如下命令

  bin\windows\kafka-server-start.bat config\server.properties

  對于Linux環境，命令如下：

  bin/kafka-server-start.sh config/server.properties

日志生成器命令：

 bin\windows\kafka-run-class.bat kafka.admin.TopicCommand --zookeeper localhost:2181 --list

在Linux上：

 bin/kafka-run-class.sh kafka.admin.TopicCommand --zookeeper localhost:2181 --list

我們將會用以下命令創建一個叫做“spark-streaming-sample-topic”的新主題：

 bin\windows\kafka-run-class.bat kafka.admin.TopicCommand --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --create --topic spark-streaming-sample-topic

在Linux上：

 bin/kafka-run-class.sh kafka.admin.TopicCommand --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --create --topic spark-streaming-sample-topic

你可以再次運行list主題命令以查看是否新主題已經被正確創建。當主題已經被創建好后，我們就可以運行日志生成器程序了。通過調用稱為SparkStreamingKafkaLogGenerator的Java類來完成此步驟。日志生成器類接收以下四個參數來指定配置參數：

• 組標識：spark-streaming-sample-group
• 主題：spark-streaming-sample-topic
• 迭代次數：50
• 間隔：1000

打開一個新的命令行窗口來運行日志生成器。我們將要為JDK、Maven和Kafka目錄分別設置三個環境變量(JAVA_HOME、MAVEN_HOME和KAFKA_HOME)。然后導航到樣例項目根目錄(比如c:\dev\projects\spark-streaming-kafka-sample-app)并運行以下命令。

mvn exec:java -Dexec.mainClass=com.sparkstreaming.kafka.example.SparkStreamingKafkaLogGenerator -Dexec.args="spark-streaming-sample-groupid spark-streaming-sample-topic 50 1000"

一旦日志生成器程序運行起來，就應該在控制臺上通過debug消息看到被創建的測試日志消息。這只是個樣例代碼，所以日志消息被隨機地生成以模擬從諸如網絡服務器這種事件源生成的持續不斷的數據流。下面的圖4展示了日志消息生產者還有正在生成的日志消息截屏。

(點擊放大圖像)

圖4.Spark流日志生成器程序輸出

Spark流命令：

• kafka_2.10-0.9.0.0.jar
• kafka-clients-0.9.0.0.jar
• metrics-core-2.2.0.jar
• spark-streaming-kafka_2.10-1.4.0.jar
• zkclient-0.3.jar

將上述JAR文件添加到classpath后我用Eclipse IDE運行了程序。日志分析Spark流程序的輸出如圖5。

(點擊放大圖像)

圖5.Spark流日志分析程序輸出

Spark流應用的可視化

(點擊放大圖像)

圖6.Spark流任務的可視化圖形

我們將要看的下一個圖形就是包含了輸入比率的流統計圖，它顯示了每秒的事件數量，以及處理所花費的毫秒數。圖7展示了Spark流程序執行期間的這些統計信息，左面是流數據還沒有產生時的情況，而右邊是數據流被發送到Kafka并且被Spark流消費者處理的情況。

圖7.為樣例程序展示流統計信息的Spark可視化

結論

Spark流庫 ，Apache Spark生態系統中的一部分，用于實時流數據的數據處理。在本文中，我們學習了如何使用Spark流API來處理由服務器日志生成的數據并基于實時數據流執行分析。

下一步是什么

參考

• 用Apache Spark進行大數據處理 -第一部分：介紹
• 用Apache Spark進行大數據處理 -第二部分：Spark SQL
• Apache Spark 主站點
• Spark流站點
• Spark流 編程指南
• Spark流 - Scala代碼示例
• Spark流 - Java代碼示例
• Data Bricks的Apache Spark 參考應用
• 使用Spark流實時地標記并處理數據 - Spark首腦2015大會 介紹
• MapR中Spark流 快速指南

關于作者

Srini Penchikala 目前是一家金融服務機構的軟件架構師，這個機構位于德克薩斯州的奧斯汀。他在軟件系統架構、設計和開發方面有超過20年的經驗。Srini目前正在撰寫一本關于NoSQL數據庫模式的書。他還是曼寧出版社出版的《 Spring Roo in Action 》一書的合著者。他還曾經出席各種會議，如JavaOne，SEI Architecture Technology Conference（SATURN），IT Architect Conference（ITARC），No Fluff Just Stuff，NoSQL Now和Project World Conference等。Srini還在InfoQ，The ServerSide，OReilly Network（ONJava），DevX Java，java.net以及JavaWorld等網站上發表過很多關于軟件系統架構、安全和風險管理以及NoSQL數據庫等方面的文章。他還是InfoQ NoSQL數據庫社區的責任編輯( http://www.infoq.com/author/Srini-Penchikala)。

查看英文原文： Big Data Processing with Apache Spark - Part 3: Spark Streaming

感謝百占輝對本文的審校。

給InfoQ中文站投稿或者參與內容翻譯工作，請郵件至editors@cn.infoq.com。也歡迎大家通過新浪微博（@InfoQ，@丁曉昀），微信（微信號： InfoQChina ）關注我們。