一直以來都想接觸Storm實時計算這塊的東西,以前在弄這個的時候,跟羣裏的一些人討論過,有的人說,直接用storm不就能夠作實時處理了,用不着那麼麻煩;其實否則,作軟件開發的都知道模塊化思想,這樣設計的緣由有兩方面:
一方面是能夠模塊化,功能劃分更加清晰,從「數據採集--數據接入--流失計算--數據輸出/存儲」
html
1).數據採集java
負責從各節點上實時採集數據,選用cloudera的flume來實現mysql
2).數據接入git
因爲採集數據的速度和數據處理的速度不必定同步,所以添加一個消息中間件來做爲緩衝,選用apache的kafkagithub
3).流式計算sql
對採集到的數據進行實時分析,選用apache的storm數據庫
4).數據輸出apache
對分析後的結果持久化,暫定用mysql編程
另外一方面是模塊化以後,加入當Storm掛掉了以後,數據採集和數據接入仍是繼續在跑着,數據不會丟失,storm起來以後能夠繼續進行流式計算;ubuntu
那麼接下來咱們來看下總體的架構圖
詳細介紹各個組件及安裝配置:
操做系統:ubuntu
Flume
Flume是Cloudera提供的一個分佈式、可靠、和高可用的海量日誌採集、聚合和傳輸的日誌收集系統,支持在日誌系統中定製各種數據發送方,用於收集數據;同時,Flume提供對數據進行簡單處理,並寫到各類數據接受方(可定製)的能力。
下圖爲flume典型的體系結構:
Flume數據源以及輸出方式:
Flume提供了從console(控制檯)、RPC(Thrift-RPC)、text(文件)、tail(UNIX tail)、syslog(syslog日誌系統,支持TCP和UDP等2種模式),exec(命令執行)等數據源上收集數據的能力,在咱們的系統中目前使用exec方式進行日誌採集。
Flume的數據接受方,能夠是console(控制檯)、text(文件)、dfs(HDFS文件)、RPC(Thrift-RPC)和syslogTCP(TCP syslog日誌系統)等。在咱們系統中由kafka來接收。
Flume下載及文檔:
Flume安裝:
複製代碼
Flume啓動命令:
複製代碼
Kafka
kafka是一種高吞吐量的分佈式發佈訂閱消息系統,她有以下特性:
kafka的目的是提供一個發佈訂閱解決方案,它能夠處理消費者規模的網站中的全部動做流數據。 這種動做(網頁瀏覽,搜索和其餘用戶的行動)是在現代網絡上的許多社會功能的一個關鍵因素。 這些數據一般是因爲吞吐量的要求而經過處理日誌和日誌聚合來解決。 對於像Hadoop的同樣的日誌數據和離線分析系統,但又要求實時處理的限制,這是一個可行的解決方案。kafka的目的是經過Hadoop的並行加載機制來統一線上和離線的消息處理,也是爲了經過集羣機來提供實時的消費。
kafka分佈式訂閱架構以下圖:--取自Kafka官網
羅寶兄弟文章上的架構圖是這樣的
其實二者沒有太大區別,官網的架構圖只是把Kafka簡潔的表示成一個Kafka Cluster,而羅寶兄弟的架構圖就相對詳細一些;
Maven,Springmvc mybatis shiro, Druid, Restful, Dubbo, ZooKeeper,Redis,FastDFS,ActiveMQ,Nginx
1. 項目核心代碼結構截圖
項目模塊依賴
特別提醒:開發人員在開發的時候能夠將本身的業務REST服務化或者Dubbo服務化
2. 項目依賴介紹
2.1 後臺管理系統、Rest服務系統、Scheculer定時調度系統依賴以下圖:
2.2 Dubbo獨立服務項目依賴以下圖:
3. 項目功能部分截圖:
zookeeper、dubbo服務啓動
dubbo管控臺
REST服務平臺
Kafka版本:0.8.0
Kafka下載及文檔:http://kafka.apache.org/
Kafka安裝:
複製代碼
啓動及測試命令:
(1) start server
複製代碼
這裏是官網上的教程,kafka自己有內置zookeeper,可是我本身在實際部署中是使用單獨的zookeeper集羣,因此第一行命令我就沒執行,這裏只是些出來給你們看下。
配置獨立的zookeeper集羣須要配置server.properties文件,講zookeeper.connect修改成獨立集羣的IP和端口
複製代碼
(2)Create a topic
複製代碼
(3)Send some messages
複製代碼
(4)Start a consumer
複製代碼
kafka-console-producer.sh和kafka-console-cousumer.sh只是系統提供的命令行工具。這裏啓動是爲了測試是否能正常生產消費;驗證流程正確性
在實際開發中仍是要自行開發本身的生產者與消費者;
kafka的安裝也能夠參考我以前寫的文章:http://blog.csdn.net/weijonathan/article/details/18075967
Storm
Twitter將Storm正式開源了,這是一個分佈式的、容錯的實時計算系統,它被託管在GitHub上,遵循 Eclipse Public License 1.0。Storm是由BackType開發的實時處理系統,BackType如今已在Twitter麾下。GitHub上的最新版本是Storm 0.5.2,基本是用Clojure寫的。
Storm的主要特色以下:
接下來重頭戲開始拉!那就是框架之間的整合啦
flume和kafka整合
1.下載
flume-kafka-plus:https://github.com/beyondj2ee/flumeng-kafka-plugin
2.提取插件中的flume-conf.properties文件
修改該文件:#source section
producer.sources.s.type = exec
producer.sources.s.command = tail -f -n+1 /mnt/hgfs/vmshare/test.log
producer.sources.s.channels = c
修改全部topic的值改成test
將改後的配置文件放進flume/conf目錄下
在該項目中提取如下jar包放入環境中flume的lib下:
注:這裏的flumeng-kafka-plugin.jar這個包,後面在github項目中已經移動到package目錄了。找不到的童鞋能夠到package目錄獲取。
完成上面的步驟以後,咱們來測試下flume+kafka這個流程有沒有走通;
咱們先啓動flume,而後再啓動kafka,啓動步驟按以前的步驟執行;接下來咱們使用kafka的kafka-console-consumer.sh腳本查看是否有flume有沒有往Kafka傳輸數據;
以上這個是個人test.log文件經過flume抓取傳到kafka的數據;說明咱們的flume和kafka流程走通了;
你們還記得剛開始咱們的流程圖麼,其中有一步是經過flume到kafka,還有一步是到hdfs的;而咱們這邊尚未提到如何存入kafka且同時存如hdfs;
flume是支持數據同步複製,同步複製流程圖以下,取自於flume官網,官網用戶指南地址:http://flume.apache.org/FlumeUserGuide.html
怎麼設置同步複製呢,看下面的配置:
複製代碼
具體配置大夥根據本身的需求去設置,這裏就不具體舉例了
kafka和storm的整合
1.下載kafka-storm0.8插件:https://github.com/wurstmeister/storm-kafka-0.8-plus
2.使用maven package進行編譯,獲得storm-kafka-0.8-plus-0.3.0-SNAPSHOT.jar包 --有轉載的童鞋注意下,這裏的包名以前寫錯了,如今改正確了!很差意思!
3.將該jar包及kafka_2.9.2-0.8.0-beta1.jar、metrics-core-2.2.0.jar、scala-library-2.9.2.jar (這三個jar包在kafka項目中能找到)
備註:若是開發的項目須要其餘jar,記得也要放進storm的Lib中好比用到了mysql就要添加mysql-connector-java-5.1.22-bin.jar到storm的lib下
那麼接下來咱們把storm也重啓下;
完成以上步驟以後,咱們還有一件事情要作,就是使用kafka-storm0.8插件,寫一個本身的Storm程序;
這裏我給大夥附上一個我弄的storm程序,百度網盤分享地址:連接: http://pan.baidu.com/s/1dD28mDr 密碼: 44r3
先稍微看下程序的建立Topology代碼
數據操做主要在WordCounter類中,這裏只是使用簡單JDBC進行插入處理
這裏只須要輸入一個參數做爲Topology名稱就能夠了!咱們這裏使用本地模式,因此不輸入參數,直接看流程是否走通;
先看下日誌,這裏打印出來了往數據庫裏面插入數據了
而後咱們查看下數據庫;插入成功了!
到這裏咱們的整個整合就完成了!
可是這裏還有一個問題,不知道大夥有沒有發現。
因爲咱們使用storm進行分佈式流式計算,那麼分佈式最須要注意的是數據一致性以及避免髒數據的產生;因此我提供的測試項目只能用於測試,正式開發不能這樣處理;
同時給的建議是創建一個zookeeper的分佈式全局鎖,保證數據一致性,避免髒數據錄入!