個人觀點:大數(shù)據(jù)我們都知道hadoop,但并不都是hadoop.我們該如何構(gòu)建大數(shù)據(jù)庫項目。對于離線處理兵睛,hadoop還是比較適合的,但是對于實時性比較強的窥浪,數(shù)據(jù)量比較大的祖很,我們可以采用Storm,那么Storm和什么技術(shù)搭配漾脂,才能夠做一個適合自己的項目假颇。下面給大家可以參考。
可以帶著下面問題來閱讀本文章:
1.一個好的項目架構(gòu)應(yīng)該具備什么特點骨稿?
2.本項目架構(gòu)是如何保證數(shù)據(jù)準確性的笨鸡?
3.什么是Kafka?
4.flume+kafka如何整合啊终?
5.使用什么腳本可以查看flume有沒有往Kafka傳輸數(shù)據(jù)
做軟件開發(fā)的都知道模塊化思想,這樣設(shè)計的原因有兩方面:
一方面是可以模塊化傲须,功能劃分更加清晰蓝牲,從“數(shù)據(jù)采集--數(shù)據(jù)接入--流失計算--數(shù)據(jù)輸出/存儲”
1).數(shù)據(jù)采集
負責從各節(jié)點上實時采集數(shù)據(jù),選用cloudera的flume來實現(xiàn)
2).數(shù)據(jù)接入
由于采集數(shù)據(jù)的速度和數(shù)據(jù)處理的速度不一定同步泰讽,因此添加一個消息中間件來作為緩沖例衍,選用apache的kafka
3).流式計算
對采集到的數(shù)據(jù)進行實時分析,選用apache的storm
4).數(shù)據(jù)輸出
對分析后的結(jié)果持久化已卸,暫定用mysql
另一方面是模塊化之后佛玄,假如當Storm掛掉了之后,數(shù)據(jù)采集和數(shù)據(jù)接入還是繼續(xù)在跑著累澡,數(shù)據(jù)不會丟失梦抢,storm起來之后可以繼續(xù)進行流式計算;
那么接下來我們來看下整體的架構(gòu)圖
詳細介紹各個組件及安裝配置:
操作系統(tǒng):ubuntu
Flume
Flume是Cloudera提供的一個分布式愧哟、可靠奥吩、和高可用的海量日志采集哼蛆、聚合和傳輸?shù)娜罩臼占到y(tǒng),支持在日志系統(tǒng)中定制各類數(shù)據(jù)發(fā)送方霞赫,用于收集數(shù)據(jù);同時腮介,F(xiàn)lume提供對數(shù)據(jù)進行簡單處理,并寫到各種數(shù)據(jù)接受方(可定制)的能力端衰。
下圖為flume典型的體系結(jié)構(gòu):
Flume數(shù)據(jù)源以及輸出方式:
Flume提供了從console(控制臺)叠洗、RPC(Thrift-RPC)、text(文件)旅东、tail(UNIX tail)灭抑、syslog(syslog日志系統(tǒng),支持TCP和UDP等2種模式)玉锌,exec(命令執(zhí)行)等數(shù)據(jù)源上收集數(shù)據(jù)的能力,在我們的系統(tǒng)中目前使用exec方式進行日志采集名挥。
Flume的數(shù)據(jù)接受方,可以是console(控制臺)主守、text(文件)禀倔、dfs(HDFS文件)、RPC(Thrift-RPC)和syslogTCP(TCP syslog日志系統(tǒng))等参淫。在我們系統(tǒng)中由kafka來接收救湖。
Flume下載及文檔:
http://flume.apache.org/
Flume安裝:
$tar zxvf apache-flume-1.4.0-bin.tar.gz/usr/local
復制代碼
Flume啟動命令:
$bin/flume-ng agent --conf conf --conf-file conf/flume-conf.properties --name producer -Dflume.root.logger=INFO,console
復制代碼
Kafka
kafka是一種高吞吐量的分布式發(fā)布訂閱消息系統(tǒng),她有如下特性:
通過O(1)的磁盤數(shù)據(jù)結(jié)構(gòu)提供消息的持久化涎才,這種結(jié)構(gòu)對于即使數(shù)以TB的消息存儲也能夠保持長時間的穩(wěn)定性能鞋既。
高吞吐量:即使是非常普通的硬件kafka也可以支持每秒數(shù)十萬的消息。
支持通過kafka服務(wù)器和消費機集群來分區(qū)消息耍铜。
支持Hadoop并行數(shù)據(jù)加載邑闺。
kafka的目的是提供一個發(fā)布訂閱解決方案,它可以處理消費者規(guī)模的網(wǎng)站中的所有動作流數(shù)據(jù)棕兼。 這種動作(網(wǎng)頁瀏覽陡舅,搜索和其他用戶的行動)是在現(xiàn)代網(wǎng)絡(luò)上的許多社會功能的一個關(guān)鍵因素。 這些數(shù)據(jù)通常是由于吞吐量的要求而通過處理日志和日志聚合來解決伴挚。 對于像Hadoop的一樣的日志數(shù)據(jù)和離線分析系統(tǒng)靶衍,但又要求實時處理的限制,這是一個可行的解決方案茎芋。kafka的目的是通過Hadoop的并行加載機制來統(tǒng)一線上和離線的消息處理颅眶,也是為了通過集群機來提供實時的消費。
kafka分布式訂閱架構(gòu)如下圖:--取自Kafka官網(wǎng)
羅寶兄弟文章上的架構(gòu)圖是這樣的
其實兩者沒有太大區(qū)別田弥,官網(wǎng)的架構(gòu)圖只是把Kafka簡潔的表示成一個Kafka Cluster涛酗,而上面架構(gòu)圖就相對詳細一些;
Kafka版本:0.8.0
Kafka下載及文檔:http://kafka.apache.org/
Kafka安裝:
> tar xzf kafka-.tgz
> cd kafka-
> ./sbt update
> ./sbt package
> ./sbt assembly-package-dependency
復制代碼
啟動及測試命令:
(1) start server
> bin/zookeeper-server-start.shconfig/zookeeper.properties
> bin/kafka-server-start.shconfig/server.properties
復制代碼
這里是官網(wǎng)上的教程,kafka本身有內(nèi)置zookeeper煤杀,但是我自己在實際部署中是使用單獨的zookeeper集群眷蜈,所以第一行命令我就沒執(zhí)行,這里只是些出來給大家看下沈自。
配置獨立的zookeeper集群需要配置server.properties文件酌儒,講zookeeper.connect修改為獨立集群的IP和端口
zookeeper.connect=nutch1:2181
復制代碼
(2)Create a topic
> bin/kafka-create-topic.sh --zookeeper localhost:2181 --replica 1 --partition 1 --topic test
> bin/kafka-list-topic.sh --zookeeperlocalhost:2181
復制代碼
(3)Send some messages
> bin/kafka-console-producer.sh--broker-list localhost:9092 --topic test
復制代碼
(4)Start a consumer
> bin/kafka-console-consumer.sh--zookeeper localhost:2181 --topic test --from-beginning
復制代碼
kafka-console-producer.sh和kafka-console-cousumer.sh只是系統(tǒng)提供的命令行工具。這里啟動是為了測試是否能正常生產(chǎn)消費枯途;驗證流程正確性
在實際開發(fā)中還是要自行開發(fā)自己的生產(chǎn)者與消費者忌怎;
kafka的安裝也可以參考我之前寫的文章:http://blog.csdn.net/weijonathan/article/details/18075967
Storm
Twitter將Storm正式開源了,這是一個分布式的酪夷、容錯的實時計算系統(tǒng)榴啸,它被托管在GitHub上,遵循??Eclipse Public License 1.0晚岭。Storm是由BackType開發(fā)的實時處理系統(tǒng)鸥印,BackType現(xiàn)在已在Twitter麾下。GitHub上的最新版本是Storm 0.5.2坦报,基本是用Clojure寫的库说。
Storm的主要特點如下:
簡單的編程模型。類似于MapReduce降低了并行批處理復雜性片择,Storm降低了進行實時處理的復雜性潜的。
可以使用各種編程語言。你可以在Storm之上使用各種編程語言字管。默認支持Clojure啰挪、Java、Ruby和Python嘲叔。要增加對其他語言的支持亡呵,只需實現(xiàn)一個簡單的Storm通信協(xié)議即可。
容錯性硫戈。Storm會管理工作進程和節(jié)點的故障锰什。
水平擴展。計算是在多個線程掏愁、進程和服務(wù)器之間并行進行的歇由。
可靠的消息處理卵牍。Storm保證每個消息至少能得到一次完整處理果港。任務(wù)失敗時,它會負責從消息源重試消息糊昙。
快速辛掠。系統(tǒng)的設(shè)計保證了消息能得到快速的處理,使用?MQ作為其底層消息隊列。(0.9.0.1版本支持?MQ和netty兩種模式)
本地模式萝衩。Storm有一個“本地模式”回挽,可以在處理過程中完全模擬Storm集群。這讓你可以快速進行開發(fā)和單元測試猩谊。
由于篇幅問題,具體的安裝步驟可以參考:Storm-0.9.0.1安裝部署 指導
接下來重頭戲開始拉墙牌!那就是框架之間的整合啦
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包放入環(huán)境中flume的lib下:
注:這里的flumeng-kafka-plugin.jar這個包暗甥,后面在github項目中已經(jīng)移動到package目錄了喜滨。找不到的童鞋可以到package目錄獲取虽风。
完成上面的步驟之后,我們來測試下flume+kafka這個流程有沒有走通寄月;
我們先啟動flume辜膝,然后再啟動kafka剥懒,啟動步驟按之前的步驟執(zhí)行初橘;接下來我們使用kafka的kafka-console-consumer.sh腳本查看是否有flume有沒有往Kafka傳輸數(shù)據(jù)保檐;
以上這個是我的test.log文件通過flume抓取傳到kafka的數(shù)據(jù)闸天;說明我們的flume和kafka流程走通了场躯;
大家還記得剛開始我們的流程圖么旅挤,其中有一步是通過flume到kafka粘茄,還有一步是到hdfs的;而我們這邊還沒有提到如何存入kafka且同時存如hdfs儒搭;
flume是支持數(shù)據(jù)同步復制搂鲫,同步復制流程圖如下,取自于flume官網(wǎng)怔檩,官網(wǎng)用戶指南地址:http://flume.apache.org/FlumeUserGuide.html
怎么設(shè)置同步復制呢,看下面的配置:
#2個channel和2個sink的配置文件??這里我們可以設(shè)置兩個sink仑氛,一個是kafka的锯岖,一個是hdfs的出吹;
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2
復制代碼
具體配置大伙根據(jù)自己的需求去設(shè)置捶牢,這里就不具體舉例了
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包--有轉(zhuǎn)載的童鞋注意下,這里的包名之前寫錯了驯耻,現(xiàn)在改正確了可缚!不好意思斋枢!
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項目中能找到)
備注:如果開發(fā)的項目需要其他jar朝扼,記得也要放進storm的Lib中比如用到了mysql就要添加mysql-connector-java-5.1.22-bin.jar到storm的lib下
那么接下來我們把storm也重啟下;
完成以上步驟之后榛斯,我們還有一件事情要做驮俗,就是使用kafka-storm0.8插件王凑,寫一個自己的Storm程序聋丝;
這里我給大伙附上一個我弄的storm程序弱睦,百度網(wǎng)盤分享地址:鏈接:http://pan.baidu.com/s/1jGBp99W密碼: 9arq
先稍微看下程序的創(chuàng)建Topology代碼
數(shù)據(jù)操作主要在WordCounter類中况木,這里只是使用簡單JDBC進行插入處理
這里只需要輸入一個參數(shù)作為Topology名稱就可以了子库!我們這里使用本地模式仑嗅,所以不輸入?yún)?shù)张症,直接看流程是否走通俗他;
storm-0.9.0.1/bin/storm jar storm-start-demo-0.0.1-SNAPSHOT.jar com.storm.topology.MyTopology
復制代碼
先看下日志兆衅,這里打印出來了往數(shù)據(jù)庫里面插入數(shù)據(jù)了
然后我們查看下數(shù)據(jù)庫;插入成功了危融!
到這里我們的整個整合就完成了吉殃!
但是這里還有一個問題蛋勺,不知道大伙有沒有發(fā)現(xiàn)鸠删。
由于我們使用storm進行分布式流式計算刃泡,那么分布式最需要注意的是數(shù)據(jù)一致性以及避免臟數(shù)據(jù)的產(chǎn)生捅僵;所以我提供的測試項目只能用于測試庙楚,正式開發(fā)不能這樣處理;
晨色星空J2EE(一個網(wǎng)名)給的建議是建立一個zookeeper的分布式全局鎖酪捡,保證數(shù)據(jù)一致性逛薇,避免臟數(shù)據(jù)錄入永罚!
zookeeper客戶端框架大伙可以使用Netflix Curator來完成呢袱,由于這塊我還沒去看翅敌,所以只能寫到這里了蚯涮!
http://blog.csdn.net/weijonathan/article/details/18301321