flume+kafka+storm運行的示例分析

這篇文章主要介紹flume+kafka+storm運行的示例分析，文中介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們一定要看完！

概述

在基于Hadoop平臺的很多應用場景中，我們需要對數據進行離線和實時分析，離線分析可以很容易地借助于Hive或者mr來實現統計分析，但是對于實時的需求Hive和mr就不合適了。實時應用場景可以使用Storm，它是一個實時處理系統，它為實時處理類應用提供了一個計算模型，可以很容易地進行編程處理。為了統一離線和實時計算，一般情況下，我們都希望將離線和實時計算的數據源的集合統一起來作為輸入，然后將數據的流向分別經由實時系統和離線分析系統，分別進行分析處理，這時我們可以考慮將數據源（如使用Flume收集日志）直接連接一個消息中間件，如Kafka，可以整合Flume+Kafka，Flume作為消息的Producer，生產的消息數據（日志數據、業務請求數據等等）發布到Kafka中，然后通過訂閱的方式，使用Storm的Topology作為消息的Consumer，在Storm集群中分別進行如下兩個需求場景的處理：
直接使用Storm的Topology對數據進行實時分析處理
整合Storm+HDFS，將消息處理后寫入HDFS進行離線分析處理

flume+kafka+storm相結合，此時，flume作為數據來源收集數據，kafka作為消息隊列，起緩沖作用，storm從kafka拉取數據分析處理。做軟件開發的都知道模塊化思想，這樣設計的原因有兩方面：
一方面是可以模塊化，功能劃分更加清晰，從“數據采集--數據接入--流式計算--數據輸出/存儲”

1）.數據采集
負責從各節點上實時采集數據，選用cloudera的flume來實現
2）.數據接入
由于采集數據的速度和數據處理的速度不一定同步，因此添加一個消息中間件來作為緩沖，選用apache的kafka
3）.流式計算
對采集到的數據進行實時分析，選用apache的storm
4）.數據輸出
對分析后的結果持久化，暫定用mysql
另一方面是模塊化之后，假如當Storm掛掉了之后，數據采集和數據接入還是繼續在跑著，數據不會丟失，storm起來之后可以繼續進行流式計算；

數據來源flume

Kafka生產的數據，是由Flume的Sink提供的，這里我們需要用到Flume集群，通過Flume集群將Agent的日志收集分發到 Kafka。我們根據情況選擇合適的source，這里我用的是exec，channel是memory，sink當然就是kafkasink。詳細配置如下：

flume到kafka

flume到kafka的傳輸過程如下圖：

kafka的配置跟之前搭建的沒有什么改動。

測試flume到kafka

flume和kafka配置好以后，先啟動flume集群，這里是后臺運行：

flume-ng agent -n agent -c /usr/local/apache-flume-1.6.0-bin/conf  -f /usr/local/apache-flume-1.6.0-bin/conf/flume-test.conf -Dflume.root.logger=DEBUG,console &

然后啟動zookeeper：

./zkServer.sh start

接著啟動kafka集群，這里是后臺運行：

./kafka-server-start.sh ../config/server.properties &

然后向監控的文件里輸入數據：

echo 'hello world' >> topic-test.txt

接著在kafka集群上創建消費者，測試flume到kafka是否聯通，當然也可以使用kafka監控工具查看：

我們可以事先創建好topic，當然我們也可以自動創建topic，設置kafka auto.create.topics.enable屬性為true，默認就為true。

./kafka-console-consumer.sh  --zookeeper master:2181 --from-beginning --topic topic1
這邊輸出'hello world'則表明flume到kafka連接成功。

 storm讀取kafka數據分析編程

首先搭建好storm集群，啟動nimbus、supervisor、ui

然后topology編程，我這里是java編程的一個小例子：

主類

package com.kafka_storm;
import java.util.HashMap;
import java.util.Map;
import storm.kafka.BrokerHosts;
import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.ZkHosts;
import storm.kafka.bolt.KafkaBolt;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.utils.Utils;
public class StormKafkaTopo {   
    public static void main(String[] args) throws Exception { 
     // 配置Zookeeper地址
        BrokerHosts brokerHosts = new ZkHosts("master:2181");
        // 配置Kafka訂閱的Topic，以及zookeeper中數據節點目錄和名字
        //這里需要注意的是，spout會根據config的后面兩個參數在zookeeper上為每個kafka分區創建保存讀取偏移的節點，如：/zkroot/topo/partition_0。
        SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, "topic1", "/zkkafkaspout" , "kafkaspout");
        
        // 配置KafkaBolt中的kafka.broker.properties(可以參考kafka java編程)
        Config conf = new Config();  
        Map<String, String> map = new HashMap<String, String>(); 
        // 配置Kafka broker地址       
        map.put("metadata.broker.list", "master:9092");
        // serializer.class為消息的序列化類
        map.put("serializer.class", "kafka.serializer.StringEncoder");
        conf.put("kafka.broker.properties", map);
        // 配置KafkaBolt生成的topic
        conf.put("topic", "topic2");
         
        //默認情況下，spout下會發射域名為bytes的binary數據，如果有需要，可以通過設置schema進行修改。
        spoutConfig.scheme = new SchemeAsMultiScheme(new MessageScheme());  
        TopologyBuilder builder = new TopologyBuilder();   
        builder.setSpout("spout", new KafkaSpout(spoutConfig));  
        builder.setBolt("bolt", new SenqueceBolt()).shuffleGrouping("spout"); 
        builder.setBolt("kafkabolt", new KafkaBolt<String, Integer>()).shuffleGrouping("bolt");        
 
        if (args != null && args.length > 0) {  
            conf.setNumWorkers(3);  
            StormSubmitter.submitTopology(args[0], conf, builder.createTopology());  
        } else {  
   
            LocalCluster cluster = new LocalCluster();  
            cluster.submitTopology("Topo", conf, builder.createTopology());  
            Utils.sleep(100000);  
            cluster.killTopology("Topo");  
            cluster.shutdown();  
        }  
    }  
}

消息處理

package com.kafka_storm;
import java.io.UnsupportedEncodingException;
import java.util.List;
import backtype.storm.spout.Scheme;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
/**
 * 使用KafkaSpout時需要子集實現Scheme接口，它主要負責從消息流中解析出需要的數據
 * @author lenovo
 *
 */
public class MessageScheme implements Scheme { 
     
    /* (non-Javadoc)
     * @see backtype.storm.spout.Scheme#deserialize(byte[])
     */
    public List<Object> deserialize(byte[] ser) {
        try {
            String msg = new String(ser, "UTF-8"); 
            return new Values(msg);
        } catch (UnsupportedEncodingException e) {  
          
        }
        return null;
    }
     
     
    /* (non-Javadoc)
     * @see backtype.storm.spout.Scheme#getOutputFields()
     */
    public Fields getOutputFields() {
        // TODO Auto-generated method stub
        return new Fields("msg");  
    }  
}

bolt

package com.kafka_storm;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
public class SenqueceBolt extends BaseBasicBolt{
     
    /* (non-Javadoc)
     * @see backtype.storm.topology.IBasicBolt#execute(backtype.storm.tuple.Tuple, backtype.storm.topology.BasicOutputCollector)
     */
    public void execute(Tuple input, BasicOutputCollector collector) {
        // TODO Auto-generated method stub
         String word = (String) input.getValue(0);  
         String out = "I'm " + word +  "!";  
         System.out.println("out=" + out);
         collector.emit(new Values(out));
    }
     
    /* (non-Javadoc)
     * @see backtype.storm.topology.IComponent#declareOutputFields(backtype.storm.topology.OutputFieldsDeclarer)
     */
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("message"));
    }
}

在集群上運行

我們要將引入的第三方包全部放到storm的lib包下面，包括kafka、zookeeper的，否則會報缺失jar包的錯

storm jar StormKafkaDemo.jar com.kafka_storm.StormKafkaTopo StormKafkaDemo

開始總體測試：

向flume監控的文件輸入數據，在storm的log日志里查看輸出，當然我們也可以在kafka里查看，因為我將結果輸出到kafka里了，topic為topic2。

日志里結果如下：

以上是“flume+kafka+storm運行的示例分析”這篇文章的所有內容，感謝各位的閱讀！希望分享的內容對大家有幫助，更多相關知識，歡迎關注億速云行業資訊頻道！