1.概述

在《Kafka實戰(zhàn)－簡單示例》一文中給大家介紹來Kafka的簡單示例，演示了如何編寫Kafka的代碼去生產(chǎn)數(shù)據(jù)和消費數(shù)據(jù)，今天給大家介紹如何去整 合一個完整的項目，本篇博客我打算為大家介紹Flume＋Kafka＋Storm的實時日志統(tǒng)計，由于涉及的內(nèi)容較多，這里先給大家梳理一個項目的運用這 些技術(shù)的流程。下面是今天的內(nèi)容目錄：

• 項目流程
• Flume
• Kafka
• Storm

下面開始今天的內(nèi)容分享。

2.項目流程

在整合這套方案的時候，項目組也是經(jīng)過一番討論，在討論中，觀點很多，有人認為直接使用Storm進行實時處理，去掉Kafka環(huán)節(jié)；也有認為直接使用Kafka的API去消費，去掉Storm的消費環(huán)節(jié)等等，但是最終組內(nèi)還是一致決定使用這套方案，原因有如下幾點：

• 業(yè)務模塊化
• 功能組件化

我們認為，Kafka在整個環(huán)節(jié)中充當?shù)穆氊煈搯我?，這項目的整個環(huán)節(jié)她就是一個中間件，下面用一個圖來說明這個原因，如下圖所示：

整個項目流程如上圖所示，這樣劃分使得各個業(yè)務模塊化，功能更加的清晰明了。

• Data Collection

負責從各個節(jié)點上實時收集用戶上報的日志數(shù)據(jù)，我們選用的是Apache的Flume NG來實現(xiàn)。

• Data Access

由于收集的數(shù)據(jù)的速度和數(shù)據(jù)處理的速度不一定是一致的，因此，這里添加了一個中間件來做處理，所使用的是Apache的Kafka，關(guān)于Kafka集群部署，大家可以參考我寫的《 Kafka實戰(zhàn)－Kafka Cluster 》。另外，有一部分數(shù)據(jù)是流向HDFS分布式文件系統(tǒng)了的，方便于為離線統(tǒng)計業(yè)務提供數(shù)據(jù)源。

• Stream Computing

在收集到數(shù)據(jù)后，我們需要對這些數(shù)據(jù)做實時處理，所選用的是Apache的Storm。關(guān)于Storm的集群搭建部署博客后面補上，較為簡單。

• Data Output

在使用Storm對數(shù)據(jù)做處理后，我們需要將處理后的結(jié)果做持久化，由于對相應速度要求較高，這里采用Redis＋MySQL來做持久化。整個項目的流程架構(gòu)圖，如下圖所示：

3.Flume

Flume是一個分布式的、高可用的海量日志收集、聚合和傳輸日志收集系統(tǒng)，支持在日志系統(tǒng)中定制各類數(shù)據(jù)發(fā)送方（如：Kafka，HDFS 等），便于收集數(shù)據(jù)。Flume提供了豐富的日志源收集類型，有：Console、RPC、Text、Tail、Syslog、Exec等數(shù)據(jù)源的收集， 在我們的日志系統(tǒng)中目前我們所使用的是spooldir方式進行日志文件采集，配置內(nèi)容信息如下所示：

producer.sources.s.type = spooldir
producer.sources.s.spoolDir = /home/hadoop/dir/logdfs

當然，F(xiàn)lume的數(shù)據(jù)發(fā)送方類型也是多種類型的，有：Console、Text、HDFS、RPC等，這里我們系統(tǒng)所使用的是Kafka中間件來接收，配置內(nèi)容如下所示：

producer.sinks.r.type = org.apache.flume.plugins.KafkaSink
producer.sinks.r.metadata.broker.list=dn1:9092,dn2:9092,dn3:9092
producer.sinks.r.partition.key=0
producer.sinks.r.partitioner.class=org.apache.flume.plugins.SinglePartition
producer.sinks.r.serializer.class=kafka.serializer.StringEncoder
producer.sinks.r.request.required.acks=0
producer.sinks.r.max.message.size=1000000
producer.sinks.r.producer.type=sync
producer.sinks.r.custom.encoding=UTF-8
producer.sinks.r.custom.topic.name=test

4.Kafka

Kafka是一種提供高吞吐量的分布式發(fā)布訂閱消息系統(tǒng)，她的特性如下所示：

• 通過磁盤數(shù)據(jù)結(jié)構(gòu)提供消息的持久化，這種結(jié)構(gòu)對于即使數(shù)據(jù)達到TB+級別的消息，存儲也能夠保持長時間的穩(wěn)定。
• 搞吞吐特性使得Kafka即使使用普通的機器硬件，也可以支持每秒數(shù)10W的消息。
• 能夠通過Kafka Cluster和Consumer Cluster來Partition消息。

Kafka的目的是提供一個發(fā)布訂閱解決方案，他可以處理Consumer網(wǎng)站中的所有流動數(shù)據(jù)，在網(wǎng)頁瀏覽，搜索以及用戶的一些行為，這些動作 是較為關(guān)鍵的因素。這些數(shù)據(jù)通常是由于吞吐量的要求而通過處理日志和日志聚合來解決。對于Hadoop這樣的日志數(shù)據(jù)和離線計算系統(tǒng)，這樣的方案是一個解 決實時處理較好的一種方案。

關(guān)于Kafka集群的搭建部署和使用，大家可以參考我寫的：《 Kafka實戰(zhàn)－Kafka Cluster 》，這里就不多做贅述了。

5.Storm

Twitter將Storm開源了，這是一個分布式的、容錯的實時計算系統(tǒng)，已被貢獻到Apache基金會，下載地址如下所示：

http://storm.apache.org/downloads.html

Storm的主要特點如下：

• 簡單的編程模型。類似于MapReduce降低了并行批處理復雜性，Storm降低了進行實時處理的復雜性。
• 可以使用各種編程語言。你可以在Storm之上使用各種編程語言。默認支持Clojure、Java、Ruby和Python。要增加對其他語言的支持，只需實現(xiàn)一個簡單的Storm通信協(xié)議即可。
• 容錯性。Storm會管理工作進程和節(jié)點的故障。
• 水平擴展。計算是在多個線程、進程和服務器之間并行進行的。
• 可靠的消息處理。Storm保證每個消息至少能得到一次完整處理。任務失敗時，它會負責從消息源重試消息。
• 快速。系統(tǒng)的設計保證了消息能得到快速的處理，使用?MQ作為其底層消息隊列。
• 本地模式。Storm有一個本地模式，可以在處理過程中完全模擬Storm集群。這讓你可以快速進行開發(fā)和單元測試。

Storm集群由一個主節(jié)點和多個工作節(jié)點組成。主節(jié)點運行了一個名為“Nimbus”的守護進程，用于分配代碼、布置任務及故障檢測。每個工作 節(jié) 點都運行了一個名為“Supervisor”的守護進程，用于監(jiān)聽工作，開始并終止工作進程。Nimbus和Supervisor都能快速失敗，而且是無 狀態(tài)的，這樣一來它們就變得十分健壯，兩者的協(xié)調(diào)工作是由Apache的ZooKeeper來完成的。

Storm的術(shù)語包括Stream、Spout、Bolt、Task、Worker、Stream Grouping和Topology。Stream是被處理的數(shù)據(jù)。Spout是數(shù)據(jù)源。Bolt處理數(shù)據(jù)。Task是運行于Spout或Bolt中的 線程。Worker是運行這些線程的進程。Stream Grouping規(guī)定了Bolt接收什么東西作為輸入數(shù)據(jù)。數(shù)據(jù)可以隨機分配（術(shù)語為Shuffle），或者根據(jù)字段值分配（術(shù)語為Fields），或者 廣播（術(shù)語為All），或者總是發(fā)給一個Task（術(shù)語為Global），也可以不關(guān)心該數(shù)據(jù)（術(shù)語為None），或者由自定義邏輯來決定（術(shù)語為 Direct）。Topology是由Stream Grouping連接起來的Spout和Bolt節(jié)點網(wǎng)絡。在Storm Concepts頁面里對這些術(shù)語有更詳細的描述。

關(guān)于Storm集群的搭建部署，博客在下一篇中更新，到時候會將更新地址附在這里，這里就先不對Storm集群的搭建部署做過多的贅述了。

6.總結(jié)

這里就是為大家介紹的Flume＋Kafka＋Storm的整體流程，后續(xù)會給大家用一個項目案例來實踐演示這個流程，包括具體的各個模塊的編碼實踐。今天大家可以先熟悉下實時計算項目的流程開發(fā)。