flume+kafka+storm+mysql架構設計

序言

前段時間學習了storm,最近剛開blog,就把這些資料放上來供你們參考。
這個框架用的組件基本都是最新穩定版本,flume-ng1.4+kafka0.8+storm0.9+mysql
(項目是maven項目,須要改動mysql配置,提供兩種topology:讀取本地文件(用來本地測試);讀取服務器日誌文件。)

架構圖

 

數據流向圖

(是visio畫的,圖太大,放上來字看起來比較小,若是有須要的朋友留郵箱)java

實時日誌分析系統架構簡介

系統主要分爲四部分:python

                       

1).數據採集mysql

負責從各節點上實時採集數據,選用cloudera的flume來實現linux

2).數據接入c++

因爲採集數據的速度和數據處理的速度不必定同步,所以添加一個消息中間件來做爲緩衝,選用apache的kafkagit

3).流式計算github

對採集到的數據進行實時分析,選用apache的stormsql

4).數據輸出apache

對分析後的結果持久化,暫定用MySQLcentos

 

詳細介紹各個組件及安裝配置:

操做系統:centos6.4

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版本:1.4.0

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

注意事項:須要更改conf目錄下的配置文件,而且添加jar包到lib目錄下。

 

Kafka

Kafka是一個消息中間件,它的特色是:

一、關注大吞吐量,而不是別的特性

二、針對實時性場景

三、關於消息被處理的狀態是在consumer端維護,而不是由kafka server端維護。

四、分佈式,producer、broker和consumer都分佈於多臺機器上。

   下圖爲kafka的架構圖:

 

 Kafka版本:0.8.0

Kafka下載及文檔:http://kafka.apache.org/

Kafka安裝:

> tar xzf kafka-<VERSION>.tgz

> cd kafka-<VERSION>

> ./sbt update

> ./sbt package

> ./sbt assembly-package-dependency Kafka

 

啓動及測試命令:

(1) start server

 > bin/zookeeper-server-start.sh config/zookeeper.properties

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

(2)Create a topic
> bin/kafka-create-topic.sh --zookeeper localhost:2181 --replica 1 --partition 1 --topic test

> bin/kafka-list-topic.sh --zookeeper localhost: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

 

storm

Storm是一個分佈式的、高容錯的實時計算系統。

Storm架構圖:

storm工做任務topology:

Storm 版本:0.9.0

Storm 下載:http://storm-project.net/

Storm安裝:

第一步,安裝Python2.7.2

# wget http://www.Python.org/ftp/python/2.7.2/Python-2.7.2.tgz

# tar zxvf Python-2.7.2.tgz

# cd Python-2.7.2

# ./configure

# make

# make install

# vi /etc/ld.so.conf

第二步,安裝zookeeper(kafka自帶zookeeper,若是選用kafka的,該步可省略)

#wget http://ftp.meisei-u.ac.jp/mirror/apache/dist//zookeeper/zookeeper-3.3.3/zoo keeper-3.3.3.tar.gz

# tar zxf zookeeper-3.3.3.tar.gz

# ln -s /usr/local/zookeeper-3.3.3/ /usr/local/zookeeper

# vi ~./bashrc (設置ZOOKEEPER_HOME和ZOOKEEPER_HOME/bin)

第三步,安裝Java

$tar zxvf  jdk-7u45-Linux-x64.tar.gz  /usr/local

 

若是使用storm0.9如下版本須要安裝zeromq及jzmq。

第四步,安裝zeromq以及jzmq

jzmq的安裝貌似是依賴zeromq的,因此應該先裝zeromq,再裝jzmq。

1)安裝zeromq(非必須):

缺乏c++環境:yum install gcc-c++

以後遇到的問題是:Error:cannot link with -luuid, install uuid-dev

這是由於沒有安裝uuid相關的package。

解決方法是:# yum install uuid*

# yum install e2fsprogs*

# yum install libuuid*

 

2)安裝jzmq(非必須)

  • # yum install git
  • # git clone git://github.com/nathanmarz/jzmq.git
  • # cd jzmq
  • # ./autogen.sh
  • # ./configure
  • # make
  • # make install

而後,jzmq就裝好了,這裏有個網站上參考到的問題沒有碰見,碰見的童鞋能夠參考下。在./autogen.sh這步若是報錯:autogen.sh:error:could not find libtool is required to run autogen.sh,這是由於缺乏了libtool,能夠用#yum install libtool*來解決。

 若是安裝的是storm0.9及以上版本不須要安裝zeromq和jzmq,可是須要修改storm.yaml來指定消息傳輸爲netty

storm.local.dir: "/tmp/storm/data"

storm.messaging.transport: "backtype.storm.messaging.netty.Context"
storm.messaging.netty.server_worker_threads: 1
storm.messaging.netty.client_worker_threads: 1
storm.messaging.netty.buffer_size: 5242880
storm.messaging.netty.max_retries: 100
storm.messaging.netty.max_wait_ms: 1000
storm.messaging.netty.min_wait_ms: 100

 

第五步,安裝storm

$unzip storm-0.9.0-wip16.zip

備註:單機版不須要修改配置文件,分佈式在修改配置文件時要注意:冒號後必須加空格。

測試storm是否安裝成功:

1. 下載strom starter的代碼 Git clone https://github.com/nathanmarz/storm-starter.git

2. 使用mvn -f m2-pom.xml package 進行編譯

若是沒有安裝過maven,參見以下步驟安裝:
1.從maven的官網下載http://maven.apache.org/

tar zxvf apache-maven-3.1.1-bin.tar.gz /usr/local

配置maven環境變量

export MAVEN_HOME=/usr/local/maven

export PATH=$PATH:$MAVEN_HOME/bin

驗證maven是否安裝成功:mvn -v

修改Storm-Starter的pom文件m2-pom.xml ,修改dependency中twitter4j-core 和 twitter4j-stream兩個包的依賴版本,以下:

 org.twitter4j
 twitter4j-core
 [2.2,)


 org.twitter4j
 twitter4j-stream
 [2.2,)

編譯完後生成target文件夾

啓動zookeeper

zkServer.sh start

啓動nimbus supervisor ui

storm nimbus

storm supervisor

storm ui

jps查看啓動狀態

進入target目錄執行:

storm jar storm-starter-0.0.1-SNAPSHOT-jar-with-dependencies.jar storm.starter.WordCountTopology wordcountTop

而後查看http://localhost:8080

註釋:單機版 不用修改storm.yaml

 

kafka和storm整合

1.下載kafka-storm0.8插件:https://github.com/wurstmeister/storm-kafka-0.8-plus

2.該項目下載下來須要調試下,找到依賴jar包。而後從新打包,做爲咱們的storm項目的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-storm-0.8-plus項目依賴中能找到)

備註:若是開發的項目須要其餘jar,記得也要放進storm的Lib中好比用到了mysql就要添加mysql-connector-java-5.1.22-bin.jar到storm的lib下

 

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下:

 

以上爲單機版的flume+kafka+storm的配置安裝

 

flume+storm插件

https://github.com/xiaochawan/edw-Storm-Flume-Connectors

 

啓動步驟

 

安裝好storm,flume,kafka以後開始項目部署啓動(在部署啓動以前最好按照安裝文檔進行storm kafka flume各個組件測試)。

1, 將編寫好的storm項目打成jar包放入服務器上,假如放在/usr/local/project/storm.xx.jar
注:關於storm項目的編寫見安裝文檔中的 kafka和storm整合 。


2, 啓動zookeeper(這裏能夠啓動kafka自帶的zookeeper或者啓動單獨安裝的kafka,如下以kafka自帶爲例)
cd /usr/local/kafka
bin/zookeeper-server-start.sh config/zookeeper.properties


3, 啓動kafka
bin/kafka-server-start.sh config/server.properties

建立主題
bin/kafka-create-topic.sh --zookeeper localhost:2181 --replica 1 --partition 1 --topic test

注:由於kafka消息的offset是由zookeeper記錄管理的,因此在此需指定zookeeper的ip,replica 表示該主題的消息被複制幾份,partition 表示每份主題被分割成幾部分。test表示主題名稱。


4. 啓動storm

> storm nimbus
> storm supervisor
> storm ui

cd /usr/local/project/
> storm jar storm.xx.jar storm.testTopology test

注:storm.xx.jar 爲咱們編寫好的storm項目jar包,第一步完成的工做。 storm.testTopology 爲storm項目中main方法所在的類路徑。test爲這次topology的名字。

5. 啓動flume

cd /usr/local/flume
bin/flume-ng agent --conf conf --conf-file conf/flume-conf.properties --name producer

注:flume.conf.properties爲咱們自定義的flume配置文件,flume安裝好後是沒有此文件的,須要咱們本身編寫,編寫方式見flume安裝的文章。

相關文章
相關標籤/搜索