Flume+Kafka+Zookeeper搭建大數(shù)據(jù)日志采集框架

1.  JDK的安裝

    參考jdk的安裝，此處略。

2.  安裝Zookeeper

    參考我的Zookeeper安裝教程中的“完全分布式”部分。

3.  安裝Kafka

    參考我的Kafka安裝教程中的“完全分布式搭建”部分。

4.  安裝Flume

    參考我的Flume安裝教程。

5.  配置Flume

5.1.  配置kafka-s.cfg

    $ cd /software/flume/conf/                   #切換到kafka的配置文件目錄

    $ cp flume-conf.properties.template kafka-s.cfg   #將flume的配置文件模板復(fù)制為kafka-s.cfg

    kafka-s.cfg的詳細(xì)內(nèi)容如下：

    #配置flume agent的source、channel、sink

    a1.sources = r1

    a1.channels = c1

    a1.sinks=k1

    #配置source

    a1.sources.r1.type = exec

    a1.sources.r1.command=tail -F/tmp/logs/kafka.log

    # 配置channel

    a1.channels.c1.type = memory

    a1.channels.c1.capacity = 1000

    a1.channels.c1.transactionCapacity = 100

    #配置sink

    a1.sinks.k1.channel = c1

    a1.sinks.k1.type =org.apache.flume.sink.kafka.KafkaSink

    #設(shè)置Kafka的Topic

    a1.sinks.k1.kafka.topic = mytest

    #設(shè)置Kafka的broker地址和端口號

    a1.sinks.k1.kafka.bootstrap.servers = s201:9092,s202:9092,s203:9092

    #配置批量提交的數(shù)量

    a1.sinks.k1.kafka.flumeBatchSize = 20

    a1.sinks.k1.kafka.producer.acks = 1

    a1.sinks.k1.kafka.producer.linger.ms = 1

    a1.sinks.ki.kafka.producer.compression.type= snappy

    # 綁定source和sink 到channel

    a1.sources.r1.channels = c1

    a1.sinks.k1.channel=c1

    關(guān)于上面配置文件中需要注意的3點：

    a、a1.sources.r1.command=tail-F /tmp/logs/kafka.log  

    b、a1.sinks.k1.kafka.bootstrap.servers= s201:9092,s202:9092,s203:9092

    c、a1.sinks.k1.topic= mytest

    由配置文件我們可以了解到：

    1）、我們需要在/tmp/logs下建一個kafka.log的文件，且向文件中輸出內(nèi)容（下面會說到）；

    2）、flume連接到kafka的地址是 s201:9092,s202:9092,s203:9092，注意不要配置錯誤了；

    3）、flume會將采集后的內(nèi)容輸出到Kafkatopic 為mytest上，所以我們啟動zk，kafka后需要打開一個終端消費topic mytest的內(nèi)容，這樣就可以看到flume與kafka之間玩起來了。

5.2.  創(chuàng)建/tmp/logs/kafka.log文件目錄及文件

在/tmp/logs下建立空文件kafka.log；如果/tmp目錄下沒有logs目錄，則需要先創(chuàng)建logs目錄。

5.3.  創(chuàng)建生成日志數(shù)據(jù)shell腳本

在hadoop用戶目錄下創(chuàng)建kafkaoutput.sh腳本，并賦予執(zhí)行權(quán)限，用來向/tmp/logs/kafka.log輸出內(nèi)容。

kafkaoutput.sh腳本的具體內(nèi)容如下：

for((i=0;i<=1000;i++));

do echo"kafka_test-"+$i>>/tmp/logs/kafka.log;

done

5.4.  啟動Zookeeper

    到zk安裝的服務(wù)器上啟動zk服務(wù)，命令如下：

    $ zkServer.sh start

5.5.  啟動Kafka

    到部署kafka每臺服務(wù)器上面啟動kafka集群

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

5.6.  創(chuàng)建Kafka Topic

    $ kafka-topics.sh --create --zookeeper s201:2181--replication-factor 3 --partitions 3 --topic mytest

5.7.  啟動消費Topic

    $ kafka-console-consumer.sh--bootstrap-server s201:9092,s202:9092,s203:9092 --zookeepers201:2181,s202:2181,s203:2181 --topic mytest --from-beginning

5.8.  啟動Flume

    $ flume-ng agent --conf/software/flume/conf/ --conf-file kafka-s.cfg --name a1-Dflume.root.logger=INFO,console

5.9.  執(zhí)行kafkaoutput.sh腳本生成日志數(shù)據(jù)

    $ ./kafkaoutput.sh

    查看日志文件內(nèi)容如下：

在kafka中查看的消費信息：