Kafka+storm+hbase - 程式人生

本部落格基於以下軟體：

Centos 7.3（1611）
kafka_2.10-0.10.2.1.tgz
zookeeper-3.4.10.tar.gz
hbase-1.3.1-bin.tar.gz
apache-storm-1.1.0.tar.gz
hadoop-2.8.0.tar.gz
jdk-8u131-linux-x64.tar.gz
IntelliJ IDEA 2017.1.3 x64

IP	role
172.17.11.85	namenode、secondarynamenode、datanode、hmaster、HRegionServer
172.17.11.86	DataNode、HRegionServer
172.17.11.87	DataNode、HRegionServer

1.首先先理一理思路，kafka–>storm

我使用一個生產者給一個固定的Topic下生產資料


public class Producer {
    private final KafkaProducer<String, String> producer;
    private final String topic;

    public Producer(String topic) {
        Properties props = new 
 Properties();
        props.put("bootstrap.servers", "172.17.11.85:9092,172.17.11.86:9092,172.17.11.87:9092");
        props.put("client.id", "DemoProducer");
        props.put("batch.size", 16384);//16M
        props.put("linger.ms", 1000);
        props.put("buffer.memory", 33554432);//32M
        props.put("key.serializer" 
, "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        producer = new KafkaProducer<>(props);
        this.topic = topic;
    }

    public void producerMsg() throws InterruptedException {
        String data = "Apache Storm is a free and open source distributed realtime computation system Storm makes it easy to reliably process unbounded streams of data doing for realtime processing what Hadoop did for batch processing. Storm is simple, can be used with any programming language, and is a lot of fun to use!\n" +
                "Storm has many use cases: realtime analytics, online machine learning, continuous computation, distributed RPC, ETL, and more. Storm is fast: a benchmark clocked it at over a million tuples processed per second per node. It is scalable, fault-tolerant, guarantees your data will be processed, and is easy to set up and operate.\n" +
                "Storm integrates with the queueing and database technologies you already use. A Storm topology consumes streams of data and processes those streams in arbitrarily complex ways, repartitioning the streams between each stage of the computation however needed. Read more in the tutorial.";
        data = data.replaceAll("[\\pP‘’“”]", "");
        String[] words = data.split(" ");
        Random _rand = new Random();

        Random rnd = new Random();
        int events = 10;
        for (long nEvents = 0; nEvents < events; nEvents++) {
            long runtime = new Date().getTime();
            int lastIPnum = rnd.nextInt(255);
            String ip = "192.168.2." + lastIPnum;
            String msg = words[_rand.nextInt(words.length)];
            try {
                producer.send(new ProducerRecord<>(topic, ip, msg));
                System.out.println("Sent message: (" + ip + ", " + msg + ")");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        Thread.sleep(10000);
    }
    public static void main(String[] args) throws InterruptedException {
        Producer producer = new Producer(Constants.TOPIC);
        producer.producerMsg();
    }
}

生產者將在兩句英文去除標點符號之後拆分為單個單詞，然後生產到執行的主題下
這應該是沒有什麼問題的，接下來就是消費者，同時又是Storm的Spout的kafkaSpout:

KafkaSpoutConfig<String, String> kafkaSpoutConfig = KafkaSpoutConfig
                    .builder(args[0], args[1])
                    .setProp(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
                    .setProp(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000)
                    .setProp(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000)
                    .setOffsetCommitPeriodMs(10000)
                    .setGroupId(args[2])
                    .setMaxUncommittedOffsets(250)
                    .setFirstPollOffsetStrategy(KafkaSpoutConfig.FirstPollOffsetStrategy.LATEST)
                    .build();



KafkaSpout<String, String> kafkaSpout = new KafkaSpout<>(kafkaSpoutConfig);

消費者（Spout）指定的主題消費資料然後發射到下一個Bolt


public class WordCountBolt extends BaseBasicBolt {
    private Map<String, Integer> counts = new HashMap<>();

    public void execute(Tuple input, BasicOutputCollector collector) {
        String level = input.getStringByField("value");
        Integer count = counts.get(level);
        if (count == null)
            count = 0;
        count++;
        counts.put(level, count);
        System.out.println("WordCountBolt Receive : "+level+"   "+count);
        collector.emit(new Values(level, count.toString()));
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word", "count"));
    }
}

2.storm->hbase

首先要從叢集中拷貝下hbase-site.xml配置檔案
這裡寫圖片描述

接下來就是API的呼叫：

 SimpleHBaseMapper Mapper = new SimpleHBaseMapper()
                    .withRowKeyField("word")
                    .withColumnFields(new Fields("count"))
                    .withColumnFamily("result");
            HBaseBolt hbaseBolt = new HBaseBolt(args[3], Mapper)
                    .withConfigKey("hbase");

3.整個拓撲的構造

builder.setSpout("kafkaSpout", kafkaSpout, 1);
//            builder.setBolt("wordSplitBolt", new WordSplitBolt(), 2)
//                    .shuffleGrouping("kafkaSpout");
            builder.setBolt("countBolt", new WordCountBolt(), 2)
                    .fieldsGrouping("kafkaSpout", new Fields("value"));
            builder.setBolt("HbaseBolt", hbaseBolt, 1)
                    .addConfiguration("hbase", new HashMap<String, Object>())
                    .shuffleGrouping("countBolt");

接下來才是真正重點的時候，重點！重點！重點！

重點1：-pom檔案的版本資訊

    <dependencies>
        <dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-core</artifactId>
            <version>1.1.0</version>
            <!--<scope>provided</scope>-->
        </dependency>
        <dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-hbase</artifactId>
            <version>1.1.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-kafka-client</artifactId>
            <version>1.1.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.7.3</version>
            <exclusions>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
    </dependencies>

對的！我匯入的是hadoop-client 2.7.3，至於為什麼呢？如果我寫成2.8.0，那將會產生以下異常

java.lang.NoSuchMethodError: org.apache.hadoop.security.authentication.util.KerberosUtil.hasKerberosTicket(Ljavax/security/auth/Subject;)Z
    at org.apache.hadoop.security.UserGroupInformation.<init>(UserGroupInformation.java:652) ~[hadoop-common-2.8.0.jar:?]
    at org.apache.hadoop.security.UserGroupInformation.loginUserFromSubject(UserGroupInformation.java:843) ~[hadoop-common-2.8.0.jar:?]
    at org.apache.hadoop.security.UserGroupInformation.getLoginUser(UserGroupInformation.java:802) ~[hadoop-common-2.8.0.jar:?]
    at org.apache.hadoop.security.UserGroupInformation.getCurrentUser(UserGroupInformation.java:675) ~[hadoop-common-2.8.0.jar:?]
    at org.apache.hadoop.hbase.security.User$SecureHadoopUser.<init>(User.java:285) ~[hbase-common-1.1.0.jar:1.1.0]
    at org.apache.hadoop.hbase.security.User$SecureHadoopUser.<init>(User.java:281) ~[hbase-common-1.1.0.jar:1.1.0]
    at org.apache.hadoop.hbase.security.User.getCurrent(User.java:185) ~[hbase-common-1.1.0.jar:1.1.0]
    at org.apache.hadoop.hbase.security.UserProvider.getCurrent(UserProvider.java:88) ~[hbase-common-1.1.0.jar:1.1.0]
    at org.apache.storm.hbase.common.HBaseClient.<init>(HBaseClient.java:43) ~[storm-hbase-1.1.0.jar:1.1.0]
    at org.apache.storm.hbase.bolt.AbstractHBaseBolt.prepare(AbstractHBaseBolt.java:75) ~[storm-hbase-1.1.0.jar:1.1.0]
    at org.apache.storm.hbase.bolt.HBaseBolt.prepare(HBaseBolt.java:109) ~[storm-hbase-1.1.0.jar:1.1.0]
    at org.apache.storm.daemon.executor$fn__5044$fn__5057.invoke(executor.clj:791) ~[storm-core-1.1.0.jar:1.1.0]
    at org.apache.storm.util$async_loop$fn__557.invoke(util.clj:482) [storm-core-1.1.0.jar:1.1.0]
    at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
    at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121]
72750 [Thread-22-HbaseBolt-executor[1 1]] ERROR o.a.s.d.executor - 
java.lang.NoSuchMethodError: org.apache.hadoop.security.authentication.util.KerberosUtil.hasKerberosTicket(Ljavax/security/auth/Subject;)Z
    at org.apache.hadoop.security.UserGroupInformation.<init>(UserGroupInformation.java:652) ~[hadoop-common-2.8.0.jar:?]
    at org.apache.hadoop.security.UserGroupInformation.loginUserFromSubject(UserGroupInformation.java:843) ~[hadoop-common-2.8.0.jar:?]
    at org.apache.hadoop.security.UserGroupInformation.getLoginUser(UserGroupInformation.java:802) ~[hadoop-common-2.8.0.jar:?]
    at org.apache.hadoop.security.UserGroupInformation.getCurrentUser(UserGroupInformation.java:675) ~[hadoop-common-2.8.0.jar:?]
    at org.apache.hadoop.hbase.security.User$SecureHadoopUser.<init>(User.java:285) ~[hbase-common-1.1.0.jar:1.1.0]
    at org.apache.hadoop.hbase.security.User$SecureHadoopUser.<init>(User.java:281) ~[hbase-common-1.1.0.jar:1.1.0]
    at org.apache.hadoop.hbase.security.User.getCurrent(User.java:185) ~[hbase-common-1.1.0.jar:1.1.0]
    at org.apache.hadoop.hbase.security.UserProvider.getCurrent(UserProvider.java:88) ~[hbase-common-1.1.0.jar:1.1.0]
    at org.apache.storm.hbase.common.HBaseClient.<init>(HBaseClient.java:43) ~[storm-hbase-1.1.0.jar:1.1.0]
    at org.apache.storm.hbase.bolt.AbstractHBaseBolt.prepare(AbstractHBaseBolt.java:75) ~[storm-hbase-1.1.0.jar:1.1.0]
    at org.apache.storm.hbase.bolt.HBaseBolt.prepare(HBaseBolt.java:109) ~[storm-hbase-1.1.0.jar:1.1.0]
    at org.apache.storm.daemon.executor$fn__5044$fn__5057.invoke(executor.clj:791) ~[storm-core-1.1.0.jar:1.1.0]
    at org.apache.storm.util$async_loop$fn__557.invoke(util.clj:482) [storm-core-1.1.0.jar:1.1.0]
    at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
    at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121]
72787 [Thread-22-HbaseBolt-executor[1 1]] ERROR o.a.s.util - Halting process: ("Worker died")
java.lang.RuntimeException: ("Worker died")
    at org.apache.storm.util$exit_process_BANG_.doInvoke(util.clj:341) [storm-core-1.1.0.jar:1.1.0]
    at clojure.lang.RestFn.invoke(RestFn.java:423) [clojure-1.7.0.jar:?]
    at org.apache.storm.daemon.worker$fn__5642$fn__5643.invoke(worker.clj:759) [storm-core-1.1.0.jar:1.1.0]
    at org.apache.storm.daemon.executor$mk_executor_data$fn__4863$fn__4864.invoke(executor.clj:274) [storm-core-1.1.0.jar:1.1.0]
    at org.apache.storm.util$async_loop$fn__557.invoke(util.clj:494) [storm-core-1.1.0.jar:1.1.0]
    at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
    at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121]

應該是版本不相容引起的

重點2：log4j-over-slf4j.jar AND slf4j-log4j12.jar衝突

SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/C:/Users/geekp/.m2/repository/org/apache/logging/log4j/log4j-slf4j-impl/2.8/log4j-slf4j-impl-2.8.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/C:/Users/geekp/.m2/repository/org/slf4j/slf4j-log4j12/1.7.10/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]


....

SLF4J: Detected both log4j-over-slf4j.jar AND slf4j-log4j12.jar on the class path, preempting StackOverflowError. 
SLF4J: See also http://www.slf4j.org/codes.html#log4jDelegationLoop for more details.

[Thread-22-HbaseBolt-executor[1 1]] ERROR o.a.s.util - Async loop died!
java.lang.NoSuchMethodError: org.apache.hadoop.security.authentication.util.KerberosUtil.hasKerberosTicket(Ljavax/security/auth/Subject;)Z
    at org.apache.hadoop.security.UserGroupInformation.<init>(UserGroupInformation.java:652) ~[hadoop-common-2.8.0.jar:?]
    at org.apache.hadoop.security.UserGroupInformation.loginUserFromSubject(UserGroupInformation.java:843) ~[hadoop-common-2.8.0.jar:?]
    at org.apache.hadoop.security.UserGroupInformation.getLoginUser(UserGroupInformation.java:802) ~[hadoop-common-2.8.0.jar:?]
    at org.apache.hadoop.security.UserGroupInformation.getCurrentUser(UserGroupInformation.java:675) ~[hadoop-common-2.8.0.jar:?]
    at org.apache.hadoop.hbase.security.User$SecureHadoopUser.<init>(User.java:285) ~[hbase-common-1.1.0.jar:1.1.0]
    at org.apache.hadoop.hbase.security.User$SecureHadoopUser.<init>(User.java:281) ~[hbase-common-1.1.0.jar:1.1.0]
    at org.apache.hadoop.hbase.security.User.getCurrent(User.java:185) ~[hbase-common-1.1.0.jar:1.1.0]
    at org.apache.hadoop.hbase.security.UserProvider.getCurrent(UserProvider.java:88) ~[hbase-common-1.1.0.jar:1.1.0]
    at org.apache.storm.hbase.common.HBaseClient.<init>(HBaseClient.java:43) ~[storm-hbase-1.1.0.jar:1.1.0]
    at org.apache.storm.hbase.bolt.AbstractHBaseBolt.prepare(AbstractHBaseBolt.java:75) ~[storm-hbase-1.1.0.jar:1.1.0]
    at org.apache.storm.hbase.bolt.HBaseBolt.prepare(HBaseBolt.java:109) ~[storm-hbase-1.1.0.jar:1.1.0]
    at org.apache.storm.daemon.executor$fn__5044$fn__5057.invoke(executor.clj:791) ~[storm-core-1.1.0.jar:1.1.0]
    at org.apache.storm.util$async_loop$fn__557.invoke(util.clj:482) [storm-core-1.1.0.jar:1.1.0]
    at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
    at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121]
71976 [Thread-22-HbaseBolt-executor[1 1]] ERROR o.a.s.d.executor - 
java.lang.NoSuchMethodError: org.apache.hadoop.security.authentication.util.KerberosUtil.hasKerberosTicket(Ljavax/security/auth/Subject;)Z
    at org.apache.hadoop.security.UserGroupInformation.<init>(UserGroupInformation.java:652) ~[hadoop-common-2.8.0.jar:?]
    at org.apache.hadoop.security.UserGroupInformation.loginUserFromSubject(UserGroupInformation.java:843) ~[hadoop-common-2.8.0.jar:?]
    at org.apache.hadoop.security.UserGroupInformation.getLoginUser(UserGroupInformation.java:802) ~[hadoop-common-2.8.0.jar:?]
    at org.apache.hadoop.security.UserGroupInformation.getCurrentUser(UserGroupInformation.java:675) ~[hadoop-common-2.8.0.jar:?]
    at org.apache.hadoop.hbase.security.User$SecureHadoopUser.<init>(User.java:285) ~[hbase-common-1.1.0.jar:1.1.0]
    at org.apache.hadoop.hbase.security.User$SecureHadoopUser.<init>(User.java:281) ~[hbase-common-1.1.0.jar:1.1.0]
    at org.apache.hadoop.hbase.security.User.getCurrent(User.java:185) ~[hbase-common-1.1.0.jar:1.1.0]
    at org.apache.hadoop.hbase.security.UserProvider.getCurrent(UserProvider.java:88) ~[hbase-common-1.1.0.jar:1.1.0]
    at org.apache.storm.hbase.common.HBaseClient.<init>(HBaseClient.java:43) ~[storm-hbase-1.1.0.jar:1.1.0]
    at org.apache.storm.hbase.bolt.AbstractHBaseBolt.prepare(AbstractHBaseBolt.java:75) ~[storm-hbase-1.1.0.jar:1.1.0]
    at org.apache.storm.hbase.bolt.HBaseBolt.prepare(HBaseBolt.java:109) ~[storm-hbase-1.1.0.jar:1.1.0]
    at org.apache.storm.daemon.executor$fn__5044$fn__5057.invoke(executor.clj:791) ~[storm-core-1.1.0.jar:1.1.0]
    at org.apache.storm.util$async_loop$fn__557.invoke(util.clj:482) [storm-core-1.1.0.jar:1.1.0]
    at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
    at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121]
71976 [Thread-26-kafkaSpout-executor[4 4]] INFO  o.a.s.k.s.KafkaSpout - Initialization complete
71992 [Thread-22-HbaseBolt-executor[1 1]] ERROR o.a.s.util - Halting process: ("Worker died")
java.lang.RuntimeException: ("Worker died")
    at org.apache.storm.util$exit_process_BANG_.doInvoke(util.clj:341) [storm-core-1.1.0.jar:1.1.0]
    at clojure.lang.RestFn.invoke(RestFn.java:423) [clojure-1.7.0.jar:?]
    at org.apache.storm.daemon.worker$fn__5642$fn__5643.invoke(worker.clj:759) [storm-core-1.1.0.jar:1.1.0]
    at org.apache.storm.daemon.executor$mk_executor_data$fn__4863$fn__4864.invoke(executor.clj:274) [storm-core-1.1.0.jar:1.1.0]
    at org.apache.storm.util$async_loop$fn__557.invoke(util.clj:494) [storm-core-1.1.0.jar:1.1.0]
    at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
    at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121]

Process finished with exit code 1

這個只需要不引入slf4j-log4j12就可以了，修改一下pom檔案：

<dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.7.3</version>
            <exclusions>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

重點3：在伺服器上的Hbase配置檔案hbase-site.xml
這個問題真的特別特別重要，md困擾了我一天

我在伺服器叢集上的配置檔案是這樣的

<configuration>
<property>
<name>hbase.rootdir</name>
<value>hdfs://master:9000/hbase</value>
</property>
<property>
<name>hbase.cluster.distributed</name>
<value>true</value>
</property>
<property>
<name>hbase.zookeeper.quorum</name>
<value>master,slave1,slave2</value>
</property>
<property>
<name>hbase.master.info.bindAddress</name>
<value>0.0.0.0</ 
 
              
           
              
              
            
            相關推薦
			   
            
            
            
 

    

    
    Kafka+storm+hbase
      
							
							
							本部落格基於以下軟體：



Centos 7.3（1611）
kafka_2.10-0.10.2.1.tgz
zookeeper-3.4.10.tar.gz
hbase-1.3.1-bin.tar.gz
apache-storm-1.1.0.tar.gz
ha 

  
 

    

    
    Kafka+Storm+HBase專案Demo(1)--CDH搭建hadoop叢集
      
							
							
							這是一個類似淘寶雙11大螢幕的專案 。

kafka+storm+hbase整合:kafka作為分散式訊息系統，實時訊息系統，有生產者和消費者；storm作為大資料的實時處理系統；hbase是apache hadoop 的資料庫，其具有高效的讀寫效能。

Clo 

  
 

    

    
    Storm Kafka + Storm + HBase例項
      
							
							
							需求


WordCount案例 Kafka + Storm + HBase
HBase表名：wordcount; 
列族：result; 
RowKey：word; 
Field：count
打包叢集部署執行




開發過程

1. 
配置kafkaSpou 

  
 

    

    
    Storm+HBASE+MySQL  實時讀取Kafka資訊計算儲存
      
								
								            
							
							
							本文是通過Storm將生產出來的資料進行實時的計算統計，整理出來之後將資料寫到hbase和mysql資料中，並將結果展示在前端頁面上，頁面展示部分在下一篇說明

題目要求

一、機組執行資料清洗規則 
 

  
 

    

    
    Flume+Kafka+Storm+Redis構建大數據實時處理系統：實時統計網站PV、UV+展示
      大數據   實時計算   Storm   [TOC]

1 大數據處理的常用方法
前面在我的另一篇文章中《大數據采集、清洗、處理：使用MapReduce進行離線數據分析完整案例》中已經有提及到，這裏依然給出下面的圖示：

前面給出的那篇文章是基於MapReduce的離線數據分析案例，其通過對網站產生的用戶訪問 

  
 

    

    
    Flume+Kafka+SparkStreaming+Hbase+可視化（一）
      日誌導入   ash   channels   style   導入   com   system   ase   spark   一、前置準備：
Linux命令基礎
Scala、Python其中一門
Hadoop、Spark、Flume、Kafka、Hbase基礎知識
 
二、分布式日誌收集框架Flume
 

  
 

    

    
    Flink操作mysql kafka和hbase
       
 
 主程式 
 package com.streaming.flink;


import java.util.Properties;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apac 

  
 

    

    
    Maven、Kafka、HBASE、flume安裝
      
                一、安裝Maven



tar -xvzf apache-maven-3.5.2-bin.tar.gz

2.新增環境變數

vim ~/.bashrc

export MAVEN_HOME=/usr/local/src/apache-maven-3.5.2
export  

  
 

    

    
    Flume+Kafka+Storm+Redis實時分析系統基本架構
       
 
 今天作者要在這裡通過一個簡單的電商網站訂單實時分析系統和大家一起梳理一下大資料環境下的實時分析系統的架構模型。當然這個架構模型只是實時分析技術的一 個簡單的入門級架構，實際生產環境中的大資料實時分析技術還涉及到很多細節的處理, 比如使用Storm的ACK機制保證資料都能被正確處理, 叢集的高可用架構 

  
 

    

    
    基於Kafka+SparkStreaming+HBase實時點選流案例
      背景Kafka實時記錄從資料採集工具Flume或業務系統實時介面收集資料，並作為訊息緩衝元件為上游實時計算框架提供可靠資料支撐，Spark 1.3版本後支援兩種整合Kafka機制（Receiver-based Approach 和 Direct Approach），具體細節請參考文章最後官方文件連結，資料儲存 

  
 

    

    
    flume+kafka+storm的整合使用
      
                
Flume-ng
Flume是一個分散式、可靠、和高可用的海量日誌採集、聚合和傳輸的系統。


       不過這裡寫寫自己的見解


這個是flume的架構圖
 從上圖可以看到幾個名詞：
Agent: 一個Agent包含Source、Channel、Sink和其他的元件 

  
 

    

    
    Flume+Kafka+Storm+Redis構建大資料實時處理系統
       
 
 資料處理方法分為離線處理和線上處理，今天寫到的就是基於Storm的線上處理。在下面給出的完整案例中，我們將會完成下面的幾項工作： 
 
  如何一步步構建我們的實時處理系統（Flume+Kafka+Storm+Redis） 
  實時處理網站的使用者訪問日誌，並統計出該網站的PV、UV 
  將實時 

  
 

    

    
    從kafka往hbase寫資料時無法寫入的問題(region server中出現與master的預設hostname相同的regionserver節點)
       
 
         在通過flume將生產的日誌傳遞到kafka再通過kafka向hbase寫入資料的時候出現問題 
 在執行到flushCommits（） 
         這一步 

  
 

    

    
    基於flume+kafka+storm日誌收集系統搭建
      
								
								            
						
                
基於flume+kafka+storm日誌收集系統搭建
1.     環境
192.168.0.2 hadoop1
192.168.0.3 hadoop2
192.168.0.4 hadoop3
已經 

  
 

    

    
    Kafka—Storm之KafkaSpout和KafkaBolt原始碼解釋
      
                

轉載來自：http://blog.csdn.net/ransom0512/article/details/50497261

另一個比較詳細的KafkaSpout詳解見：http://www.cnblogs.com/cruze/p/4241181.html


Storm 

  
 

    

    
    flume-ng+Kafka+Storm+HDFS 實時系統搭建
      
                
今天搭建了一下storm流處理系統,整個搭建的流程都是參考以下地址:http://www.cnblogs.com/web-v/articles/6510090.html
文章中並沒有給出flume同時寫入kafka和hdfs時的配置檔案。以下是我的flume配置檔案,有一些 

  
 

    

    
    flume-ng+Kafka+Storm+HDFS 實時系統組合
      
                

大資料我們都知道hadoop，但並不都是hadoop.我們該如何構建大資料庫專案。對於離線處理，hadoop還是比較適合的，但是對於實時性比較強的，資料量比較大的，我們可以採用Storm，那麼Storm和什麼技術搭配，才能夠做一個適合自己的專案。下面給大家可以參考。

可 

  
 

    

    
    flume讀取日誌資料寫入kafka   然後kafka+storm整合
      
                

一、flume配置

flume要求1.6以上版本

flume-conf.properties檔案配置內容，sinks的輸出作為kafka的product






a1.sources = r1
a1.sinks = k1
a1.cha 

  
 

    

    
    分散式訊息中介軟體（四）——Flume+Kafka+Storm+Redis生態架構實戰
      
                
一、Kafka專案應用架構分析
1、Kafka生態架構
 
     資料收集的速度，跟處理的速度不一定一致，故使用Kafka中介軟體作為資料收集和資料處理的一個Access入口，接收flume收集的資料，並通過kafkaSpout提交給Storm進行處理。
2、kafka 

  
 

    

    
    新版flume+kafka+storm安裝部署
      Download the
 0.8.2.0 release and un-tar it.
> tar -xzf kafka_2.10-0.8.2.0.tgz
> cd kafka_2.10-0.8.2.0

Step 2: Start the server
Kafka uses ZooKeeper