基于Hadoop生態(tài)SparkStreaming的大數(shù)據(jù)實(shí)時(shí)流處理平臺(tái)的搭建
隨著公司業(yè)務(wù)發(fā)展,對(duì)大數(shù)據(jù)的獲取和實(shí)時(shí)處理的要求就會(huì)越來(lái)越高,日志處理、用戶行為分析、場(chǎng)景業(yè)務(wù)分析等等,傳統(tǒng)的寫日志方式根本滿足不了業(yè)務(wù)的實(shí)時(shí)處理需求,所以本人準(zhǔn)備開始著手改造原系統(tǒng)中的數(shù)據(jù)處理方式,重新搭建一個(gè)實(shí)時(shí)流處理平臺(tái),主要是基于hadoop生態(tài),利用Kafka作為中轉(zhuǎn),SparkStreaming框架實(shí)時(shí)獲取數(shù)據(jù)并清洗,將結(jié)果多維度的存儲(chǔ)進(jìn)HBase數(shù)據(jù)庫(kù)。
整個(gè)平臺(tái)大致的框架如下:

操作系統(tǒng):Centos7
用到的框架:
- Flume1.8.0
- Hadoop2.9.0
- kafka2.11-1.0.0
- Spark2.2.1
- HBase1.2.6
- ZooKeeper3.4.11
- maven3.5.2
整體的開發(fā)環(huán)境是基于JDK1.8以上以及Scala,所以得提前把java和Scala的環(huán)境給準(zhǔn)備好,接下來(lái)就開始著手搭建基礎(chǔ)平臺(tái):
一、配置開發(fā)環(huán)境
下載并解壓JDK1.8,、下載并解壓Scala,配置profile文件:

二、配置zookeeper、maven環(huán)境
下載并解壓zookeeper以及maven并配置profile文件

然后配置一下zoo.cfg里面的相關(guān)配置,指定一下dataDir目錄等等
啟動(dòng)zookeeper:
- /usr/local/zookeeper-3.4.11/bin/zkServer.sh start
如果不報(bào)錯(cuò),jps看一下是否啟動(dòng)成功
三、安裝配置Hadoop
Hadoop的安裝配置在之前文章中有說(shuō)過(guò)(傳送門),為了下面的步驟方便理解,這里只做一個(gè)單機(jī)版的簡(jiǎn)單配置說(shuō)明:
下載hadoop解壓并配置環(huán)境:

首先配置hadoop-env.sh、yarn-env.sh,修改JAVA_HOME到指定的JDK安裝目錄/usr/local/java/jdk1.8.0_144
創(chuàng)建hadoop的工作目錄
- mkdir /opt/data/hadoop
編輯core-site.xml、hdfs-site.xml、yarn-site.xml等相關(guān)配置文件,具體配置不再闡述請(qǐng)看前面的文章,配置完成之后記得執(zhí)行hadoop namenode -format,否則hdfs啟動(dòng)會(huì)報(bào)錯(cuò),啟動(dòng)完成后不出問(wèn)題瀏覽器訪問(wèn)50070端口會(huì)看到hadoop的頁(yè)面。
四、安裝配置kafka
還是一樣,先下載kafka,然后配置:

進(jìn)入kafka的config目錄,配置server.properties,指定log.dirs和zookeeper.connect參數(shù);配置zookeeper.properties文件中zookeeper的dataDir,配置完成后啟動(dòng)kafka
- kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties
可以用jps查看有沒(méi)有kafka進(jìn)程,然后測(cè)試一下kafka是否能夠正常收發(fā)消息,開兩個(gè)終端,一個(gè)用來(lái)做producer發(fā)消息一個(gè)用來(lái)做consumer收消息,首先,先創(chuàng)建一個(gè)topic

如果不出一下會(huì)看到如下輸出:

然后在第一個(gè)終端中輸入命令:
- kafka-console-producer.sh –broker-list localhost:9092 –topic testTopic
在第二個(gè)終端中輸入命令:
- kafka-console-consumer.sh –zookeeper 127.0.0.1:2181 –topic testTopic
如果啟動(dòng)都正常,那么這兩個(gè)終端將進(jìn)入阻塞監(jiān)聽狀態(tài),在第一個(gè)終端中輸入任何消息第二個(gè)終端都將會(huì)接收到。
五、安裝配置HBase
下載并解壓HBase:

修改hbase下的配置文件,首先修改hbase-env.sh,主要修改JAVA_HOME以及相關(guān)參數(shù),這里要說(shuō)明一下HBASE_MANAGES_ZK這個(gè)參數(shù),因?yàn)椴捎昧俗约旱膠ookeeper,所以這里設(shè)置為false,否則hbase會(huì)自己?jiǎn)?dòng)一個(gè)zookeeper

然后修改hbase-site.xml,我們?cè)O(shè)置hbase的文件放在hdfs中,所以要設(shè)置hdfs地址,其中tsk1是我安裝hadoop的機(jī)器的hostname,hbase.zookeeper.quorum參數(shù)是安裝zookeeper的地址,這里的各種地址最好用機(jī)器名

配置完成后啟動(dòng)hbase,輸入命令:
- start-hbase.sh
完成后查看日志沒(méi)有報(bào)錯(cuò)的話測(cè)試一下hbase,用hbase shell進(jìn)行測(cè)試:

至此,hbase搭建成功,訪問(wèn)以下hadoop的頁(yè)面,查看file system(菜單欄Utilities->Browse the file system),這時(shí)可以看見base的相關(guān)文件已經(jīng)載hadoop的文件系統(tǒng)中。

六、安裝spark
下載spark并解壓

七、測(cè)試
至此,環(huán)境基本搭建完成,以上搭建的環(huán)境僅是服務(wù)器生產(chǎn)環(huán)境的一部分,涉及服務(wù)器信息、具體調(diào)優(yōu)信息以及集群的搭建就不寫在這里了,下面我們寫一段代碼整體測(cè)試一下從kafka生產(chǎn)消息到spark streaming接收到,然后處理消息并寫入HBase。先寫一個(gè)HBase的連接類HBaseHelper:


再寫一個(gè)測(cè)試類KafkaRecHbase用來(lái)做spark-submit提交


編譯提交到服務(wù)器,執(zhí)行命令:

沒(méi)報(bào)錯(cuò)的話執(zhí)行kafka的producer,輸入幾行數(shù)據(jù)在HBase內(nèi)就能看到結(jié)果了!
八、裝一個(gè)Flume實(shí)時(shí)采集Nginx日志寫入Kafka
Flume是一個(gè)用來(lái)日志采集的框架,安裝和配置都比較簡(jiǎn)單,可以支持多個(gè)數(shù)據(jù)源和輸出,具體可以參考Flume的文檔,寫的比較全 傳送門
下載Flume并配置環(huán)境

寫一個(gè)Flume的配置文件在flume的conf目錄下:

kafka創(chuàng)建一個(gè)名為flumeKafka的topic用來(lái)接收,然后啟動(dòng)flume:

如果沒(méi)有報(bào)錯(cuò),F(xiàn)lume將開始采集opt/data/nginxLog/nginxLog.log中產(chǎn)生的日志并實(shí)時(shí)推送給kafka,再按照上面方法寫一個(gè)spark streaming的處理類進(jìn)行相應(yīng)的處理就好。
OK!全部搞定,然而~~~~就這樣就搞定了?NO!!!這只是萬(wàn)里長(zhǎng)征的第一步!呵呵!