2、Flume1.7.0入门：安装、部署、及flume的案例
一、什么是Flume?
flume 作為 cloudera 開發(fā)的實(shí)時(shí)日志收集系統(tǒng)，受到了業(yè)界的認(rèn)可與廣泛應(yīng)用。
flume的特點(diǎn)：
flume是一個(gè)分布式、可靠、和高可用的海量日志采集、聚合和傳輸?shù)南到y(tǒng)。支持在日志系統(tǒng)中定制各類數(shù)據(jù)發(fā)送方，用于收集數(shù)據(jù);同時(shí)，Flume提供對(duì)數(shù)據(jù)進(jìn)行簡(jiǎn)單處理，并寫到各種數(shù)據(jù)接受方(比如文本、HDFS、Hbase等)的能力 。
flume的數(shù)據(jù)流由事件(Event)貫穿始終。事件是Flume的基本數(shù)據(jù)單位，它攜帶日志數(shù)據(jù)(字節(jié)數(shù)組形式)并且攜帶有頭信息，這些Event由Agent外部的Source生成，當(dāng)Source捕獲事件后會(huì)進(jìn)行特定的格式化，然后Source會(huì)把事件推入(單個(gè)或多個(gè))Channel中。你可以把Channel看作是一個(gè)緩沖區(qū)，它將保存事件直到Sink處理完該事件。Sink負(fù)責(zé)持久化日志或者把事件推向另一個(gè)Source。
flume的可靠性?
當(dāng)節(jié)點(diǎn)出現(xiàn)故障時(shí)，日志能夠被傳送到其他節(jié)點(diǎn)上而不會(huì)丟失。Flume提供了三種級(jí)別的可靠性保障，從強(qiáng)到弱依次分別為：end-to-end（收到數(shù)據(jù)agent首先將event寫到磁盤上，當(dāng)數(shù)據(jù)傳送成功后，再刪除；如果數(shù)據(jù)發(fā)送失敗，可以重新發(fā)送。），Store on failure（這也是scribe采用的策略，當(dāng)數(shù)據(jù)接收方crash時(shí)，將數(shù)據(jù)寫到本地，待恢復(fù)后，繼續(xù)發(fā)送），Besteffort（數(shù)據(jù)發(fā)送到接收方后，不會(huì)進(jìn)行確認(rèn)）。
flume的可恢復(fù)性：
還是靠Channel。推薦使用FileChannel，事件持久化在本地文件系統(tǒng)里(性能較差)。?
flume的一些核心概念：
- Agent:使用JVM 運(yùn)行Flume。每臺(tái)機(jī)器運(yùn)行一個(gè)agent，但是可以在一個(gè)agent中包含多個(gè)sources和sinks。
- Client:生產(chǎn)數(shù)據(jù)，運(yùn)行在一個(gè)獨(dú)立的線程。
- Source:從Client專門用來(lái)收集數(shù)據(jù)，傳遞給Channel，可以處理各種類型、各種格式的日志數(shù)據(jù),包括avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy、自定義。
- Sink:從Channel收集數(shù)據(jù)，運(yùn)行在一個(gè)獨(dú)立線程,sink組件是用于把數(shù)據(jù)發(fā)送到目的地的組件，目的地包括hdfs、logger、avro、thrift、ipc、file、null、Hbase、solr、自定義。?
- Channel:連接 sources 和 sinks ，這個(gè)有點(diǎn)像一個(gè)隊(duì)列,source組件把數(shù)據(jù)收集來(lái)以后，臨時(shí)存放在channel中，即channel組件在agent中是專門用來(lái)存放臨時(shí)數(shù)據(jù)的——對(duì)采集到的數(shù)據(jù)進(jìn)行簡(jiǎn)單的緩存，可以存放在memory、jdbc、file等等。
- Events:可以是日志記錄、 avro 對(duì)象等。
Agent的概念
Flume以agent為最小的獨(dú)立運(yùn)行單位。一個(gè)agent就是一個(gè)JVM,agent本身是一個(gè)Java進(jìn)程，運(yùn)行在日志收集節(jié)點(diǎn)—所謂日志收集節(jié)點(diǎn)就是服務(wù)器節(jié)點(diǎn)。?
單agent由Source、Sink和Channel三大組件構(gòu)成，類似生產(chǎn)者、倉(cāng)庫(kù)、消費(fèi)者的架構(gòu).如下圖：
Event的概念?
flume的核心是把數(shù)據(jù)從數(shù)據(jù)源(source)收集過來(lái)，在將收集到的數(shù)據(jù)送到指定的目的地(sink)。為了保證輸送的過程一定成功，在送到目的地(sink)之前，會(huì)先緩存數(shù)據(jù)(channel),待數(shù)據(jù)真正到達(dá)目的地(sink)后，flume在刪除自己緩存的數(shù)據(jù)。?
在整個(gè)數(shù)據(jù)的傳輸?shù)倪^程中，流動(dòng)的是event，即事務(wù)保證是在event級(jí)別進(jìn)行的。那么什么是event呢？—–event將傳輸?shù)臄?shù)據(jù)進(jìn)行封裝，是flume傳輸數(shù)據(jù)的基本單位，如果是文本文件，通常是一行記錄，event也是事務(wù)的基本單位。event從source，流向channel，再到sink，本身為一個(gè)字節(jié)數(shù)組，并可攜帶headers(頭信息)信息。event代表著一個(gè)數(shù)據(jù)的最小完整單元，從外部數(shù)據(jù)源來(lái)，向外部的目的地去。?
為了方便大家理解，給出一張event的數(shù)據(jù)流向圖：?
一個(gè)完整的event包括：event headers、event body、event信息(即文本文件中的單行記錄)，如下所以：?
2017-03-29 14:00:58,227 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 68 65 6C 6C 6F 20 77 6F 72 64 ? ? ? ? ? ? ? ? ? hello word }
其中event信息就是flume收集到的日記記錄。?
flume的運(yùn)行機(jī)制?
flume的核心就是一個(gè)agent，這個(gè)agent對(duì)外有兩個(gè)進(jìn)行交互的地方，一個(gè)是接受數(shù)據(jù)的輸入——source，一個(gè)是數(shù)據(jù)的輸出sink，sink負(fù)責(zé)將數(shù)據(jù)發(fā)送到外部指定的目的地。source接收到數(shù)據(jù)之后，將數(shù)據(jù)發(fā)送給channel，chanel作為一個(gè)數(shù)據(jù)緩沖區(qū)會(huì)臨時(shí)存放這些數(shù)據(jù)，隨后sink會(huì)將channel中的數(shù)據(jù)發(fā)送到指定的地方—-例如HDFS等，注意：只有在sink將channel中的數(shù)據(jù)成功發(fā)送出去之后，channel才會(huì)將臨時(shí)數(shù)據(jù)進(jìn)行刪除，這種機(jī)制保證了數(shù)據(jù)傳輸?shù)目煽啃耘c安全性。?
flume的廣義用法?
flume可以支持多級(jí)flume的agent，即flume可以前后相繼，例如sink可以將數(shù)據(jù)寫到下一個(gè)agent的source中，這樣的話就可以連成串了，可以整體處理了。flume還支持扇入(fan-in)、扇出(fan-out)。所謂扇入就是source可以接受多個(gè)輸入，所謂扇出就是sink可以將數(shù)據(jù)輸出多個(gè)目的地destination中。?
值得注意的是，Flume提供了大量?jī)?nèi)置的Source、Channel和Sink類型。不同類型的Source,Channel和Sink可以自由組合。組合方式基于用戶設(shè)置的配置文件，非常靈活。比如：Channel可以把事件暫存在內(nèi)存里，也可以持久化到本地硬盤上。Sink可以把日志寫入HDFS, HBase，甚至是另外一個(gè)Source等等。Flume支持用戶建立多級(jí)流，也就是說(shuō)，多個(gè)agent可以協(xié)同工作，并且支持Fan-in、Fan-out、Contextual Routing、Backup Routes。如下圖所示:
二、安裝Flume
1、下載Flume
http://apache.mirrors.hoobly.com/flume/1.7.0/apache-flume-1.7.0-bin.tar.gz?
2、安裝Flume
1)將下載的flume包，解壓到/opt目錄中.
2)修改 flume-env.sh 配置文件,主要是JAVA_HOME變量設(shè)置
mbp:apache-flume-1.7.0-bin$ cp conf/flume-env.sh.template? conf/flume-env.sh
mbp:apache-flume-1.7.0-bin$ vi conf/flume-env.sh
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.? See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.? The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.? You may obtain a copy of the License at
#
# ? ? http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# If this file is placed at FLUME_CONF_DIR/flume-env.sh, it will be sourced
# during Flume startup.
# Enviroment variables can be set here.
export JAVA_HOME=/Library/Java/JavaVirtualMachines/jdk1.8.0_111.jdk/Contents/Home
# Give Flume more memory and pre-allocate, enable remote monitoring via JMX
# export JAVA_OPTS="-Xms100m -Xmx2000m -Dcom.sun.management.jmxremote"
# Let Flume write raw event data and configuration information to its log files for debugging
# purposes. Enabling these flags is not recommended in production,
# as it may result in logging sensitive user information or encryption secrets.
# export JAVA_OPTS="$JAVA_OPTS -Dorg.apache.flume.log.rawdata=true -Dorg.apache.flume.log.printconfig=true "?
# Note that the Flume conf directory is always included in the classpath.
#FLUME_CLASSPATH=""
3)驗(yàn)證是否安裝成功
mbp:apache-flume-1.7.0-bin$ bin/flume-ng version
Flume 1.7.0
Source code repository: https://git-wip-us.apache.org/repos/asf/flume.git
Revision: 511d868555dd4d16e6ce4fedc72c2d1454546707
Compiled by bessbd on Wed Oct 12 20:51:10 CEST 2016
From source with checksum 0d21b3ffdc55a07e1d08875872c00523
mbp:apache-flume-1.7.0-bin$?
出現(xiàn)上面的信息，表示安裝成功了.
三、flume的案例
對(duì)于flume的原理其實(shí)很容易理解，我們更應(yīng)該掌握f(shuō)lume的具體使用方法，flume提供了大量?jī)?nèi)置的Source、Channel和Sink類型。而且不同類型的Source、Channel和Sink可以自由組合—–組合方式基于用戶設(shè)置的配置文件，非常靈活。比如：Channel可以把事件暫存在內(nèi)存里，也可以持久化到本地硬盤上。Sink可以把日志寫入HDFS, HBase，甚至是另外一個(gè)Source等等。下面我將用具體的案例詳述flume的具體用法。?
其實(shí)flume的用法很簡(jiǎn)單—-書寫一個(gè)配置文件，在配置文件當(dāng)中描述source、channel與sink的具體實(shí)現(xiàn)，而后運(yùn)行一個(gè)agent實(shí)例，在運(yùn)行agent實(shí)例的過程中會(huì)讀取配置文件的內(nèi)容，這樣flume就會(huì)采集到數(shù)據(jù)。?
配置文件的編寫原則：
0)、案例1：Netcat?
NetCat Source：監(jiān)聽一個(gè)指定的網(wǎng)絡(luò)端口，即只要應(yīng)用程序向這個(gè)端口里面寫數(shù)據(jù)，這個(gè)source組件就可以獲取到信息。
1>從整體上描述代理agent中sources、sinks、channels所涉及到的組件??
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
2>詳細(xì)描述agent中每一個(gè)source、sink與channel的具體實(shí)現(xiàn)：即在描述source的時(shí)候，需要指定source到底是什么類型的，即這個(gè)source是接受文件的、還是接受http的、還是接受thrift的；對(duì)于sink也是同理，需要指定結(jié)果是輸出到HDFS中，還是Hbase中啊等等；對(duì)于channel需要指定是內(nèi)存啊，還是數(shù)據(jù)庫(kù)啊，還是文件啊等等。
# Describe configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
3>通過channel將source與sink連接起來(lái)
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
啟動(dòng)agent的shell操作：
flume-ng agent -n a1 -c conf -f conf/example.file -Dflume.root.logger=DEBUG,console
參數(shù)說(shuō)明：
-n 指定agent名稱(與配置文件中代理的名字相同)?
-c 指定flume中配置文件的目錄?
-f 指定配置文件?
-Dflume.root.logger=DEBUG,console 設(shè)置日志等級(jí)
1)、案例1：Avro
Avro可以發(fā)送一個(gè)給定的文件給Flume，Avro 源使用AVRO RPC機(jī)制。
a)創(chuàng)建agent配置文件
mbp:apache-flume-1.7.0-bin$ vi conf/avro.conf?
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe configure the source
a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 4141
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
b)啟動(dòng)flume agent a1
mbp:apache-flume-1.7.0-bin$ bin/flume-ng agent -n a1 -c conf -f conf/avro.conf -Dflume.root.logger=INFO,console
c)創(chuàng)建指定文件
mbp:apache-flume-1.7.0-bin$ echo "hello word" > log.00
d)使用avro-client發(fā)送文件
mbp:apache-flume-1.7.0-bin$ bin/flume-ng avro-client -c conf -H mbp -p 4141 -F log.00
e)在mbp的控制臺(tái)，可以看到以下信息，注意最后一行：
2017-03-29 13:52:19,139 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.node.PollingPropertiesFileConfigurationProvider.start(PollingPropertiesFileConfigurationProvider.java:62)] Configuration provider starting
.
.
.
2017-03-29 14:00:58,227 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 68 65 6C 6C 6F 20 77 6F 72 64 ? ? ? ? ? ? ? ? ? hello word }
2)、案例2：Spool
監(jiān)聽一個(gè)指定的目錄，即只要應(yīng)用程序向這個(gè)指定的目錄中添加新的文件，source組件就可以獲取到該信息，并解析該文件的內(nèi)容，然后寫入到channle。寫入完成后，標(biāo)記該文件已完成或者刪除該文件。
Spool監(jiān)測(cè)配置的目錄下新增的文件，并將文件中的數(shù)據(jù)讀取出來(lái)。需要注意兩點(diǎn)：
1) 拷貝到spool目錄下的文件不可以再打開編輯。
2) spool目錄下不可包含相應(yīng)的子目錄
a)創(chuàng)建agent配置文件
mbp:apache-flume-1.7.0-bin$ vi conf/spool.conf
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = spooldir
a1.sources.r1.channels = c1
a1.sources.r1.spoolDir = /opt/apache-flume-1.7.0-bin/logs
a1.sources.r1.fileHeader = true
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
b)啟動(dòng)flume agent a1
mbp:apache-flume-1.7.0-bin$ bin/flume-ng agent -n a1 -c conf -f conf/spool.conf -Dflume.root.logger=INFO,console
c)追加文件到/opt/apache-flume-1.7.0-bin/logs目錄
mbp:apache-flume-1.7.0-bin$ echo "spool test1" > logs/spool_text.log
d)在mbp的控制臺(tái)，可以看到以下相關(guān)信息：
2017-03-29 14:31:04,921 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{file=/opt/apache-flume-1.7.0-bin/logs/spool_text.log} body: 73 70 6F 6F 6C 20 74 65 73 74 31? ? ? ? ? ? ? ? spool test1 }
3)、案例3：Exec
監(jiān)聽一個(gè)指定的命令，獲取一條命令的結(jié)果作為它的數(shù)據(jù)源?
常用的是tail -F file指令，即只要應(yīng)用程序向日志(文件)里面寫數(shù)據(jù)，source組件就可以獲取到日志(文件)中最新的內(nèi)容 。
EXEC執(zhí)行一個(gè)給定的命令獲得輸出的源,如果要使用tail命令，必選使得file足夠大才能看到輸出內(nèi)容
a)創(chuàng)建agent配置文件
mbp:apache-flume-1.7.0-bin$ vi conf/exec_tail.conf
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.channels = c1
a1.sources.r1.command = tail -F /opt/apache-flume-1.7.0-bin/log_exec_tail
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
b)啟動(dòng)flume agent a1
mbp:apache-flume-1.7.0-bin$ bin/flume-ng agent -n a1 -c conf -f conf/exec_tail.conf -Dflume.root.logger=INFO,console
c)生成足夠多的內(nèi)容在文件里
mbp:apache-flume-1.7.0-bin$ for i in {1..100};do echo "exec tail$i" >> /opt/apache-flume-1.7.0-bin/logs/log_exec_tail done
e)在mbp的控制臺(tái)，可以看到以下信息：
2017-03-29 15:26:25,990 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.node.PollingPropertiesFileConfigurationProvider.start(PollingPropertiesFileConfigurationProvider.java:62)] Configuration provider starting
.
.
.
2017-03-29 15:26:44,336 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 39 34? ? ? ? ? ? ? ? exec tail94 }
2017-03-29 15:26:44,336 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 39 35? ? ? ? ? ? ? ? exec tail95 }
2017-03-29 15:26:44,336 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 39 36? ? ? ? ? ? ? ? exec tail96 }
2017-03-29 15:26:44,336 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 39 37? ? ? ? ? ? ? ? exec tail97 }
2017-03-29 15:26:44,336 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 39 38? ? ? ? ? ? ? ? exec tail98 }
2017-03-29 15:26:44,337 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 39 39? ? ? ? ? ? ? ? exec tail99 }
2017-03-29 15:26:44,337 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 31 30 30 ? ? ? ? ? ? exec tail100 }
4)、案例4：Syslogtcp
Syslogtcp監(jiān)聽TCP的端口做為數(shù)據(jù)源?
a)創(chuàng)建agent配置文件
mbp:apache-flume-1.7.0-bin$ vi conf/syslog_tcp.conf
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 5140
a1.sources.r1.host = localhost
a1.sources.r1.channels = c1
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
b)啟動(dòng)flume agent a1
mbp:apache-flume-1.7.0-bin$ bin/flume-ng agent -n a1 -c conf -f conf/syslog_tcp.conf -Dflume.root.logger=INFO,console
c)測(cè)試產(chǎn)生syslog
mbp:apache-flume-1.7.0-bin$ echo "hello idoall.org syslog" | nc localhost 5140
d)在mbp的控制臺(tái)，可以看到以下信息：
2017-03-29 15:33:43,305 (New I/O worker #1) [WARN - org.apache.flume.source.SyslogUtils.buildEvent(SyslogUtils.java:317)] Event created from Invalid Syslog data.
2017-03-29 15:33:46,303 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{Severity=0, Facility=0, flume.syslog.status=Invalid} body: 68 65 6C 6C 6F 20 69 64 6F 61 6C 6C 2E 6F 72 67 hello idoall.org }
5)、案例5：JSONHandler
a)創(chuàng)建agent配置文件
mbp:apache-flume-1.7.0-bin$ vi conf/post_json.conf
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = org.apache.flume.source.http.HTTPSource
a1.sources.r1.port = 8888
a1.sources.r1.channels = c1
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
b)啟動(dòng)flume agent a1
mbp:apache-flume-1.7.0-bin$ bin/flume-ng agent -n a1 -c conf -f conf/post_json.conf -Dflume.root.logger=INFO,console
c)生成JSON 格式的POST request
curl -X POST -d '[{ "headers" :{"a" : "a1","b" : "b1"},"body" : "idoall.org_body"}]' http://localhost:8888
d)在mbp的控制臺(tái)，可以看到以下信息：
2017-03-29 15:37:30,565 (lifecycleSupervisor-1-0) [INFO - org.mortbay.log.Slf4jLog.info(Slf4jLog.java:67)] jetty-6.1.26
2017-03-29 15:37:30,713 (lifecycleSupervisor-1-0) [INFO - org.mortbay.log.Slf4jLog.info(Slf4jLog.java:67)] Started SelectChannelConnector@0.0.0.0:8888
2017-03-29 15:37:30,713 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.instrumentation.MonitoredCounterGroup.register(MonitoredCounterGroup.java:119)] Monitored counter group for type: SOURCE, name: r1: Successfully registered new MBean.
2017-03-29 15:37:30,713 (lifecycleSupervisor-1-0) [INFO - org.apache.flume.instrumentation.MonitoredCounterGroup.start(MonitoredCounterGroup.java:95)] Component type: SOURCE, name: r1 started
2017-03-29 15:38:00,451 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{a=a1, b=b1} body: 69 64 6F 61 6C 6C 2E 6F 72 67 5F 62 6F 64 79? ? idoall.org_body }?
總結(jié)Exec source：Exec source和Spooling Directory Source是兩種常用的日志采集的方式，其中Exec source可以實(shí)現(xiàn)對(duì)日志的實(shí)時(shí)采集，Spooling Directory Source在對(duì)日志的實(shí)時(shí)采集上稍有欠缺，盡管Exec source可以實(shí)現(xiàn)對(duì)日志的實(shí)時(shí)采集，但是當(dāng)Flume不運(yùn)行或者指令執(zhí)行出錯(cuò)時(shí)，Exec source將無(wú)法收集到日志數(shù)據(jù)，日志會(huì)出現(xiàn)丟失，從而無(wú)法保證收集日志的完整性。
6)、案例6：Avro Source
監(jiān)聽一個(gè)指定的Avro 端口，通過Avro 端口可以獲取到Avro client發(fā)送過來(lái)的文件 。即只要應(yīng)用程序通過Avro 端口發(fā)送文件，source組件就可以獲取到該文件中的內(nèi)容。 其中 Sink：hdfs Channel：file?
(注：Avro和Thrift都是一些序列化的網(wǎng)絡(luò)端口–通過這些網(wǎng)絡(luò)端口可以接受或者發(fā)送信息，Avro可以發(fā)送一個(gè)給定的文件給Flume，Avro 源使用AVRO RPC機(jī)制)?
Avro Source運(yùn)行原理如下圖：
flume配置文件的書寫是相當(dāng)靈活的—-不同類型的Source、Channel和Sink可以自由組合！
最后對(duì)上面用的幾個(gè)flume source進(jìn)行適當(dāng)總結(jié)：?
① NetCat Source：監(jiān)聽一個(gè)指定的網(wǎng)絡(luò)端口，即只要應(yīng)用程序向這個(gè)端口里面寫數(shù)據(jù)，這個(gè)source組件?
就可以獲取到信息。?
②Spooling Directory Source：監(jiān)聽一個(gè)指定的目錄，即只要應(yīng)用程序向這個(gè)指定的目錄中添加新的文?
件，source組件就可以獲取到該信息，并解析該文件的內(nèi)容，然后寫入到channle。寫入完成后，標(biāo)記?
該文件已完成或者刪除該文件。?
③Exec Source：監(jiān)聽一個(gè)指定的命令，獲取一條命令的結(jié)果作為它的數(shù)據(jù)源?
常用的是tail -F file指令，即只要應(yīng)用程序向日志(文件)里面寫數(shù)據(jù)，source組件就可以獲取到日志(文件)中最新的內(nèi)容 。?
④Avro Source：監(jiān)聽一個(gè)指定的Avro 端口，通過Avro 端口可以獲取到Avro client發(fā)送過來(lái)的文件 。即只要應(yīng)用程序通過Avro 端口發(fā)送文件，source組件就可以獲取到該文件中的內(nèi)容。
7)、案例7：Hadoop sink
a)創(chuàng)建agent配置文件
mbp:apache-flume-1.7.0-bin$ vi conf/hdfs_sink.conf
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 5140
a1.sources.r1.host = localhost
a1.sources.r1.channels = c1
# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.channel = c1
a1.sinks.k1.hdfs.path = hdfs://localhost:8020/user/flume/syslogtcp
a1.sinks.k1.hdfs.filePrefix = Syslog
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
b)啟動(dòng)flume agent a1
mbp:apache-flume-1.7.0-bin$ bin/flume-ng agent -n a1 -c conf -f conf/hdfs_sink.conf -Dflume.root.logger=INFO,console
c)測(cè)試產(chǎn)生syslog
mbp:apache-flume-1.7.0-bin$ echo "hello idoall flume -> hadoop testing one" | nc localhost 5140
d)在mbp的控制臺(tái)，可以看到以下信息：
2017-03-29 19:10:14,820 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.hdfs.HDFSSequenceFile.configure(HDFSSequenceFile.java:63)] writeFormat = Writable, UseRawLocalFileSystem = false
2017-03-29 19:10:14,834 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.hdfs.BucketWriter.open(BucketWriter.java:231)] Creating hdfs://localhost:8020/user/flume/syslogtcp/Syslog.1490785814821.tmp
2017-03-29 19:10:44,861 (hdfs-k1-roll-timer-0) [INFO - org.apache.flume.sink.hdfs.BucketWriter.close(BucketWriter.java:357)] Closing hdfs://localhost:8020/user/flume/syslogtcp/Syslog.1490785814821.tmp
2017-03-29 19:10:44,880 (hdfs-k1-call-runner-9) [INFO - org.apache.flume.sink.hdfs.BucketWriter$8.call(BucketWriter.java:618)] Renaming hdfs://localhost:8020/user/flume/syslogtcp/Syslog.1490785814821.tmp to hdfs://localhost:8020/user/flume/syslogtcp/Syslog.1490785814821
2017-03-29 19:10:44,884 (hdfs-k1-roll-timer-0) [INFO - org.apache.flume.sink.hdfs.HDFSEventSink$1.run(HDFSEventSink.java:382)] Writer callback called.
e)在mbp上再打開一個(gè)窗口，去hadoop上檢查文件是否生成
mbp:hadoop-2.7.3$ bin/hadoop fs -ls /user/flume/syslogtcp
Found 1 items
-rw-r--r-- ? 3 liudebin supergroup? ? ? ? 175 2017-03-29 19:10 /user/flume/syslogtcp/Syslog.1490785779051
mbp:hadoop-2.7.3$ bin/hadoop fs -cat /user/flume/syslogtcp/Syslog.1490785779051
SEQ!org.apache.hadoop.io.LongWritable"org.apache.hadoop.io.BytesWritable?L??????? ??[???(hello idoall flume -> hadoop testing one?????L??????? ??
?8)、案例8：File Roll Sink
a)創(chuàng)建agent配置文件
mbp:apache-flume-1.7.0-bin$ vi conf/file_roll.conf
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 5555
a1.sources.r1.host = localhost
a1.sources.r1.channels = c1
# Describe the sink
a1.sinks.k1.type = file_roll
a1.sinks.k1.sink.directory = /opt/apache-flume-1.7.0-bin/logs
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
b)啟動(dòng)flume agent a1
mbp:apache-flume-1.7.0-bin$ bin/flume-ng agent -n a1 -c conf -f conf/file_roll.conf -Dflume.root.logger=INFO,console
c)測(cè)試產(chǎn)生log
mbp:apache-flume-1.7.0-bin$ echo "hello idoall.org syslog" | nc localhost 5555
mbp:apache-flume-1.7.0-bin$ echo "hello idoall.org syslog 2" | nc localhost 5555
d)查看/home/hadoop/flume-1.5.0-bin/logs下是否生成文件,默認(rèn)每30秒生成一個(gè)新文件
mbp:apache-flume-1.7.0-bin$ ls -l logs/
total 24
-rw-r--r--? 1 liudebin? wheel? ? 50? 3 30 13:06 1490850370723-1
-rw-r--r--? 1 liudebin? wheel ? ? 0? 3 30 13:06 1490850370723-2
-rw-r--r--? 1 liudebin? wheel? 6429? 3 29 14:06 flume.log.COMPLETED
mbp:apache-flume-1.7.0-bin$ cat logs/1490850370723-1 logs/1490850370723-2
hello idoall.org syslog
hello idoall.org syslog 2
9)、案例9：Replicating Channel Selector
Flume支持Fan out流從一個(gè)源到多個(gè)通道。有兩種模式的Fan out，分別是復(fù)制和復(fù)用。在復(fù)制的情況下，流的事件被發(fā)送到所有的配置通道。在復(fù)用的情況下，事件被發(fā)送到可用的渠道中的一個(gè)子集。Fan out流需要指定源和Fan out通道的規(guī)則。
這次我們需要用到mbp1,mbp2兩臺(tái)機(jī)器
a)在mbp1創(chuàng)建replicating_Channel_Selector.conf配置文件
mbp1:apache-flume-1.7.0-bin$ vi conf/replicating_Channel_Selector.conf
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2
# Describe/configure the source
a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 5140
a1.sources.r1.host = localhost
a1.sources.r1.channels = c1 c2
a1.sources.r1.selector.type = replicating
# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = mbp1
a1.sinks.k1.port = 5555
a1.sinks.k2.type = avro
a1.sinks.k2.channel = c2
a1.sinks.k2.hostname = mbp2
a1.sinks.k2.port = 5555
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.channels.c2.type = memory
a1.channels.c2.capacity = 1000
a1.channels.c2.transactionCapacity = 100
b)在mbp1創(chuàng)建replicating_Channel_Selector_avro.conf配置文件
mbp1:apache-flume-1.7.0-bin$ vi conf/replicating_Channel_Selector_avro.conf
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 5555
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
c)在mbp1上將2個(gè)配置文件復(fù)制到mbp2上一份
mbp1:apache-flume-1.7.0-bin$ scp -r conf/replicating_Channel_Selector.conf vagrant@mbp2:/opt/apache-flume-1.7.0/conf/
mbp1:apache-flume-1.7.0-bin$ scp -r conf/replicating_Channel_Selector_avro.conf vagrant@mbp2:/opt/apache-flume-1.7.0/conf/
d)打開4個(gè)窗口，在mbp1和mbp2上同時(shí)啟動(dòng)兩個(gè)flume agent
mbp1:apache-flume-1.7.0-bin$ bin/flume-ng agent -c . -f conf/replicating_Channel_Selector.conf -n a1 -Dflume.root.logger=INFO,console
mbp1:apache-flume-1.7.0-bin$ bin/flume-ng agent -c . -f conf/replicating_Channel_Selector_avro.conf -n a1 -Dflume.root.logger=INFO,console
e)然后在mbp1或mbp2的任意一臺(tái)機(jī)器上，測(cè)試產(chǎn)生syslog
mbp1:apache-flume-1.7.0-bin$ echo "hello idoall.org syslog" | nc localhost 5140
f)在mbp1和mbp2的sink窗口，分別可以看到以下信息,這說(shuō)明信息得到了同步：
17/04/05 14:08:18 INFO ipc.NettyServer: Connection to /192.168.1.51:46844 disconnected.
17/04/05 14:08:52 INFO ipc.NettyServer: [id: 0x90f8fe1f, /192.168.1.50:35873 => /192.168.1.50:5555] OPEN
17/04/05 14:08:52 INFO ipc.NettyServer: [id: 0x90f8fe1f, /192.168.1.50:35873 => /192.168.1.50:5555] BOUND: /192.168.1.50:5555
17/04/05 14:08:52 INFO ipc.NettyServer: [id: 0x90f8fe1f, /192.168.1.50:35873 => /192.168.1.50:5555] CONNECTED: /192.168.1.50:35873
17/04/05 14:08:59 INFO ipc.NettyServer: [id: 0xd6318635, /192.168.1.51:46858 => /192.168.1.50:5555] OPEN
17/04/05 14:08:59 INFO ipc.NettyServer: [id: 0xd6318635, /192.168.1.51:46858 => /192.168.1.50:5555] BOUND: /192.168.1.50:5555
17/04/05 14:08:59 INFO ipc.NettyServer: [id: 0xd6318635, /192.168.1.51:46858 => /192.168.1.50:5555] CONNECTED: /192.168.1.51:46858
17/04/05 14:09:20 INFO sink.LoggerSink: Event: { headers:{Severity=0, flume.syslog.status=Invalid, Facility=0} body: 68 65 6C 6C 6F 20 69 64 6F 61 6C 6C 2E 6F 72 67 hello idoall.org }
10)、案例10：Multiplexing Channel Selector
a)在mbp1創(chuàng)建Multiplexing_Channel_Selector配置文件
mbp:apache-flume-1.7.0-bin$ vi conf/Multiplexing_Channel_Selector.conf
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2
# Describe/configure the source
a1.sources.r1.type = org.apache.flume.source.http.HTTPSource
a1.sources.r1.port = 5140
a1.sources.r1.channels = c1 c2
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = type
#映射允許每個(gè)值通道可以重疊。默認(rèn)值可以包含任意數(shù)量的通道。
a1.sources.r1.selector.mapping.baidu = c1
a1.sources.r1.selector.mapping.ali = c2
a1.sources.r1.selector.default = c1
# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = mbp1
a1.sinks.k1.port = 5555
a1.sinks.k2.type = avro
a1.sinks.k2.channel = c2
a1.sinks.k2.hostname = mbp2
a1.sinks.k2.port = 5555
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.channels.c2.type = memory
a1.channels.c2.capacity = 1000
a1.channels.c2.transactionCapacity = 100
b)在mbp1創(chuàng)建Multiplexing_Channel_Selector_avro配置文件
mbp:apache-flume-1.7.0-bin$ vi conf/Multiplexing_Channel_Selector_avro.conf
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 5555
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
c)將2個(gè)配置文件復(fù)制到mbp2上一份
mbp1:apache-flume-1.7.0-bin$scp -r conf/Multiplexing_Channel_Selector.conf??root@mbp2:/opt/apache-flume-1.7.0-bin/conf/Multiplexing_Channel_Selector.conf
mbp1:apache-flume-1.7.0-bin$scp -r conf/Multiplexing_Channel_Selector_avro.conf?root@mbp2:/opt/apache-flume-1.7.0-bin/conf/Multiplexing_Channel_Selector_avro.conf
d)打開4個(gè)窗口，在mbp1和mbp2上同時(shí)啟動(dòng)兩個(gè)flume agent
mbp1:apache-flume-1.7.0-bin$bin/flume-ng agent -c . -f conf/Multiplexing_Channel_Selector_avro.conf -n a1 -Dflume.root.logger=INFO,console
mbp1:apache-flume-1.7.0-bin$bin/flume-ng agent -c . -f conf/Multiplexing_Channel_Selector.conf -n a1 -Dflume.root.logger=INFO,console
e)然后在mbp1或mbp2的任意一臺(tái)機(jī)器上，測(cè)試產(chǎn)生syslog
mbp1:apache-flume-1.7.0-bin$curl -X POST -d '[{ "headers" :{"type" : "baidu"},"body" : "idoall_TEST1"}]'?http://localhost:5140?&& curl -X POST -d '[{ "headers" :{"type" : "ali"},"body" : "idoall_TEST2"}]'?http://localhost:5140?&& curl -X POST -d '[{ "headers" :{"type" : "qq"},"body" : "idoall_TEST3"}]'?http://localhost:5140
f)在mbp1的sink窗口，可以看到以下信息：
17/04/05 14:32:21 INFO node.Application: Starting Sink k1
17/04/05 14:32:21 INFO node.Application: Starting Source r1
17/04/05 14:32:21 INFO source.AvroSource: Starting Avro source r1: { bindAddress: 0.0.0.0, port: 5555 }...
17/04/05 14:32:21 INFO instrumentation.MonitoredCounterGroup: Monitored counter group for type: SOURCE, name: r1: Successfully registered new MBean.
17/04/05 14:32:21 INFO instrumentation.MonitoredCounterGroup: Component type: SOURCE, name: r1 started
17/04/05 14:32:21 INFO source.AvroSource: Avro source r1 started.
17/04/05 14:32:36 INFO ipc.NettyServer: [id: 0xcf00eea6, /192.168.1.50:35916 => /192.168.1.50:5555] OPEN
17/04/05 14:32:36 INFO ipc.NettyServer: [id: 0xcf00eea6, /192.168.1.50:35916 => /192.168.1.50:5555] BOUND: /192.168.1.50:5555
17/04/05 14:32:36 INFO ipc.NettyServer: [id: 0xcf00eea6, /192.168.1.50:35916 => /192.168.1.50:5555] CONNECTED: /192.168.1.50:35916
17/04/05 14:32:44 INFO ipc.NettyServer: [id: 0x432f5468, /192.168.1.51:46945 => /192.168.1.50:5555] OPEN
17/04/05 14:32:44 INFO ipc.NettyServer: [id: 0x432f5468, /192.168.1.51:46945 => /192.168.1.50:5555] BOUND: /192.168.1.50:5555
17/04/05 14:32:44 INFO ipc.NettyServer: [id: 0x432f5468, /192.168.1.51:46945 => /192.168.1.50:5555] CONNECTED: /192.168.1.51:46945
17/04/05 14:34:11 INFO sink.LoggerSink: Event: { headers:{type=baidu} body: 69 64 6F 61 6C 6C 5F 54 45 53 54 31 idoall_TEST1 }
17/04/05 14:34:57 INFO sink.LoggerSink: Event: { headers:{type=qq} body: 69 64 6F 61 6C 6C 5F 54 45 53 54 33
g)在mbp2的sink窗口，可以看到以下信息：
17/04/05 14:32:27 INFO node.Application: Starting Sink k1
17/04/05 14:32:27 INFO node.Application: Starting Source r1
17/04/05 14:32:27 INFO source.AvroSource: Starting Avro source r1: { bindAddress: 0.0.0.0, port: 5555 }...
17/04/05 14:32:27 INFO instrumentation.MonitoredCounterGroup: Monitored counter group for type: SOURCE, name: r1: Successfully registered new MBean.
17/04/05 14:32:27 INFO instrumentation.MonitoredCounterGroup: Component type: SOURCE, name: r1 started
17/04/05 14:32:27 INFO source.AvroSource: Avro source r1 started.
17/04/05 14:32:36 INFO ipc.NettyServer: [id: 0x7c2f0aec, /192.168.1.50:38104 => /192.168.1.51:5555] OPEN
17/04/05 14:32:36 INFO ipc.NettyServer: [id: 0x7c2f0aec, /192.168.1.50:38104 => /192.168.1.51:5555] BOUND: /192.168.1.51:5555
17/04/05 14:32:36 INFO ipc.NettyServer: [id: 0x7c2f0aec, /192.168.1.50:38104 => /192.168.1.51:5555] CONNECTED: /192.168.1.50:38104
17/04/05 14:32:44 INFO ipc.NettyServer: [id: 0x3d36f553, /192.168.1.51:48599 => /192.168.1.51:5555] OPEN
17/04/05 14:32:44 INFO ipc.NettyServer: [id: 0x3d36f553, /192.168.1.51:48599 => /192.168.1.51:5555] BOUND: /192.168.1.51:5555
17/04/05 14:32:44 INFO ipc.NettyServer: [id: 0x3d36f553, /192.168.1.51:48599 => /192.168.1.51:5555] CONNECTED: /192.168.1.51:48599
17/04/05 14:34:33 INFO sink.LoggerSink: Event: { headers:{type=ali} body: 69 64 6F 61 6C 6C 5F 54 45 53 54 32 idoall_TEST2 }
可以看到，根據(jù)header中不同的條件分布到不同的channel上
11)、案例11：Flume Sink Processors
failover的機(jī)器是一直發(fā)送給其中一個(gè)sink，當(dāng)這個(gè)sink不可用的時(shí)候，自動(dòng)發(fā)送到下一個(gè)sink。
a)在m1創(chuàng)建Flume_Sink_Processors配置文件
b)在m1創(chuàng)建Flume_Sink_Processors_avro配置文件
c)將2個(gè)配置文件復(fù)制到m2上一份
d)打開4個(gè)窗口，在m1和m2上同時(shí)啟動(dòng)兩個(gè)flume agent
e)然后在m1或m2的任意一臺(tái)機(jī)器上，測(cè)試產(chǎn)生log
f)因?yàn)閙2的優(yōu)先級(jí)高，所以在m2的sink窗口，可以看到以下信息，而m1沒有：
g)這時(shí)我們停止掉m2機(jī)器上的sink(ctrl+c)，再次輸出測(cè)試數(shù)據(jù)：
h)可以在m1的sink窗口，看到讀取到了剛才發(fā)送的兩條測(cè)試數(shù)據(jù)：
i)我們?cè)僭趍2的sink窗口中，啟動(dòng)sink：
j)輸入兩批測(cè)試數(shù)據(jù)：
k)在m2的sink窗口，我們可以看到以下信息，因?yàn)閮?yōu)先級(jí)的關(guān)系，log消息會(huì)再次落到m2上：
?
12)、案例12：Load balancing Sink Processor
load balance type和failover不同的地方是，load balance有兩個(gè)配置，一個(gè)是輪詢，一個(gè)是隨機(jī)。兩種情況下如果被選擇的sink不可用，就會(huì)自動(dòng)嘗試發(fā)送到下一個(gè)可用的sink上面。
a)在m1創(chuàng)建Load_balancing_Sink_Processors配置文件
mbp:apache-flume-1.7.0-bin$ vi conf/load_balancing_sink_processors.conf
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1
#這個(gè)是配置Load balancing的關(guān)鍵，需要有一個(gè)sink group
a1.sinkgroups = g1
a1.sinkgroups.g1.sinks = k1 k2
a1.sinkgroups.g1.processor.type = load_balance
a1.sinkgroups.g1.processor.backoff = true
a1.sinkgroups.g1.processor.selector = round_robin
# Describe/configure the source
a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 5140
a1.sources.r1.channels = c1
# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = m1
a1.sinks.k1.port = 5555
a1.sinks.k2.type = avro
a1.sinks.k2.channel = c1
a1.sinks.k2.hostname = m2
a1.sinks.k2.port = 5555
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
b)在m1創(chuàng)建Load_balancing_Sink_Processors_avro配置文件
mbp:apache-flume-1.7.0-bin$ vi conf/load_balancing_sink_processors_avro.conf
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 5555
# Describe the sink
sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
c)將2個(gè)配置文件復(fù)制到mbp2上一份
root@mbp1:/opt/apache-flume-1.7.0-bin# scp -r conf/Load_balancing_Sink_Processors.conf??root@mbp2:/opt/apache-flume-1.7.0-bin/conf/Load_balancing_Sink_Processors.conf
root@mbp1:/opt/apache-flume-1.7.0-bin# scp -r conf/Load_balancing_Sink_Processors_avro.conf?root@mbp2:/opt/apache-flume-1.7.0-bin/conf/Load_balancing_Sink_Processors_avro.conf
d)打開4個(gè)窗口，在mbp1和mbp2上同時(shí)啟動(dòng)兩個(gè)flume agent
root@mbp1:/opt# /opt/apache-flume-1.7.0-bin/bin/flume-ng agent -c . -f conf/Load_balancing_Sink_Processors_avro.conf -n a1 -Dflume.root.logger=INFO,console
root@mbp1:/opt# /opt/apache-flume-1.7.0-bin/bin/flume-ng agent -c . -f conf/Load_balancing_Sink_Processors.conf -n a1 -Dflume.root.logger=INFO,console
e)然后在mbp1或mbp2的任意一臺(tái)機(jī)器上，測(cè)試產(chǎn)生log，一行一行輸入，輸入太快，容易落到一臺(tái)機(jī)器上
root@mbp1:/opt# echo "idoall.org test1" | nc localhost 5140
root@mbp1:/opt# echo "idoall.org test2" | nc localhost 5140
root@mbp1:/opt# echo "idoall.org test3" | nc localhost 5140
root@mbp1:/opt# echo "idoall.org test4" | nc localhost 5140
f)在mbp1的sink窗口，可以看到以下信息：
17/04/05 15:35:29 INFO sink.LoggerSink: Event: { headers:{Severity=0, flume.syslog.status=Invalid, Facility=0} body: 69 64 6F 61 6C 6C 2E 6F 72 67 20 74 65 73 74 32 idoall.org test2 }
17/04/05 15:35:33 INFO sink.LoggerSink: Event: { headers:{Severity=0, flume.syslog.status=Invalid, Facility=0} body: 69 64 6F 61 6C 6C 2E 6F 72 67 20 74 65 73 74 34 idoall.org test4 }
g)在mbp2的sink窗口，可以看到以下信息：
17/04/05 15:35:27 INFO sink.LoggerSink: Event: { headers:{Severity=0, flume.syslog.status=Invalid, Facility=0} body: 69 64 6F 61 6C 6C 2E 6F 72 67 20 74 65 73 74 31 idoall.org test1 }
17/08/05 15:35:29 INFO sink.LoggerSink: Event: { headers:{Severity=0, flume.syslog.status=Invalid, Facility=0} body: 69 64 6F 61 6C 6C 2E 6F 72 67 20 74 65 73 74 33 idoall.org test3 }
說(shuō)明輪詢模式起到了作用。
13)、案例13：Hbase sink
a)在測(cè)試之前，請(qǐng)先參考《ubuntu12.04+hadoop2.2.0+zookeeper3.4.5+hbase0.96.2+hive0.13.1分布式環(huán)境部署》將hbase啟動(dòng)
b)然后將以下文件復(fù)制到flume中：
cp hbase-1.3.0-bin/lib/protobuf-java-2.5.0.jar apache-flume-1.7.0-bin/lib/
cp hbase-1.3.0-bin/lib/habase-client-1.3.0.jar apache-flume-1.7.0-bin/lib/
cp hbase-1.3.0-bin/lib/hbase-common-1.3.0.jar apache-flume-1.7.0-bin/lib/
cp hbase-1.3.0-bin/lib/hbase-protocol-1.3.0.jar apache-flume-1.7.0-bin/lib/
cp hbase-1.3.0-bin/lib/hbase-server-1.3.0.jar apache-flume-1.7.0-bin/lib/
cp hbase-1.3.0-bin/lib/hbase-hadoop2-compat-1.3.0.jar apache-flume-1.7.0-bin/lib/
cp hbase-1.3.0-bin/lib/hbase-hadoop-compat-1.3.0.jar apache-flume-1.7.0-bin/lib/
cp hbase-1.3.0-bin/lib/htrace-core-3.1.0-incubating.jar apache-flume-1.7.0-bin/lib/
c)確保test_idoall_org表在hbase中已經(jīng)存在
d)在m1創(chuàng)建hbase_simple配置文件
mbp:apache-flume-1.7.0-bin$ vi conf/hbase_simple.conf
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 5140
a1.sources.r1.host = localhost
a1.sources.r1.channels = c1
# Describe the sink
a1.sinks.k1.type = logger
a1.sinks.k1.type = hbase
a1.sinks.k1.table = test_idoall_org
a1.sinks.k1.columnFamily = name
a1.sinks.k1.column = idoall
a1.sinks.k1.serializer =??org.apache.flume.sink.hbase.RegexHbaseEventSerializer
a1.sinks.k1.channel = memoryChannel
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
e)啟動(dòng)flume agent
mbp:apache-flume-1.7.0-bin$ bin/flume-ng agent -n a1 -c conf -f conf/hbase_simple.conf -Dflume.root.logger=INFO,console
f)測(cè)試產(chǎn)生syslog
mbp:apache-flume-1.7.0-bin$ echo "hello idoall.org from flume" | nc localhost 5140
g)這時(shí)登錄到hbase中，可以發(fā)現(xiàn)新數(shù)據(jù)已經(jīng)插入
mbp@opt# hbase-1.3.0/bin/hbase shell
2017-04-05 16:09:48,984 INFO [main] Configuration.deprecation: hadoop.native.lib is deprecated. Instead, use io.native.lib.available
HBase Shell; enter 'help<RETURN>' for list of supported commands.
Type "exit<RETURN>" to leave the HBase Shell
Version 0.96.2-hadoop2, r1581096, Mon Mar 24 16:03:18 PDT 2017
hbase(main):001:0> list
TABLE
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/hbase-1.3.0/lib/slf4j-log4j12-1.6.4.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/hadoop-2.8.0/share/hadoop/common/lib/slf4j-log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
hbase2hive_idoall
hive2hbase_idoall
test_idoall_org
3 row(s) in 2.6880 seconds
=> ["hbase2hive_idoall", "hive2hbase_idoall", "test_idoall_org"]
hbase(main):002:0> scan "test_idoall_org"
ROW COLUMN+CELL
10086 column=name:idoall, timestamp=1406424831473, value=idoallvalue
1 row(s) in 0.0550 seconds
hbase(main):003:0> scan "test_idoall_org"
ROW COLUMN+CELL
10086 column=name:idoall, timestamp=1406424831473, value=idoallvalue
1407658495588-XbQCOZrKK8-0 column=name:payload, timestamp=1407658498203, value=hello idoall.org from flume
2 row(s) in 0.0200 seconds
hbase(main):004:0> quit
經(jīng)過這么多flume的例子測(cè)試，如果你全部做完后，會(huì)發(fā)現(xiàn)flume的功能真的很強(qiáng)大，可以進(jìn)行各種搭配來(lái)完成你想要的工作，俗話說(shuō)師傅領(lǐng)進(jìn)門，修行在個(gè)人，如何能夠結(jié)合你的產(chǎn)品業(yè)務(wù)，將flume更好的應(yīng)用起來(lái)，快去動(dòng)手實(shí)踐吧。
?
?
?
?
?
轉(zhuǎn)載于:https://www.cnblogs.com/netbloomy/p/6666683.html
總結(jié)
以上是生活随笔為你收集整理的2、Flume1.7.0入门：安装、部署、及flume的案例的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 安装spring-tool-suite插
- 下一篇： (转) Docker EE/Docker