2、Flume1.7.0入门：安装、部署、及flume的案例
一、什么是Flume?
flume 作為 cloudera 開發的實時日志收集系統，受到了業界的認可與廣泛應用。
flume的特點：
flume是一個分布式、可靠、和高可用的海量日志采集、聚合和傳輸的系統。支持在日志系統中定制各類數據發送方，用于收集數據;同時，Flume提供對數據進行簡單處理，并寫到各種數據接受方(比如文本、HDFS、Hbase等)的能力 。
flume的數據流由事件(Event)貫穿始終。事件是Flume的基本數據單位，它攜帶日志數據(字節數組形式)并且攜帶有頭信息，這些Event由Agent外部的Source生成，當Source捕獲事件后會進行特定的格式化，然后Source會把事件推入(單個或多個)Channel中。你可以把Channel看作是一個緩沖區，它將保存事件直到Sink處理完該事件。Sink負責持久化日志或者把事件推向另一個Source。
flume的可靠性?
當節點出現故障時，日志能夠被傳送到其他節點上而不會丟失。Flume提供了三種級別的可靠性保障，從強到弱依次分別為：end-to-end（收到數據agent首先將event寫到磁盤上，當數據傳送成功后，再刪除；如果數據發送失敗，可以重新發送。），Store on failure（這也是scribe采用的策略，當數據接收方crash時，將數據寫到本地，待恢復后，繼續發送），Besteffort（數據發送到接收方后，不會進行確認）。
flume的可恢復性：
還是靠Channel。推薦使用FileChannel，事件持久化在本地文件系統里(性能較差)。?
flume的一些核心概念：
- Agent:使用JVM 運行Flume。每臺機器運行一個agent，但是可以在一個agent中包含多個sources和sinks。
- Client:生產數據，運行在一個獨立的線程。
- Source:從Client專門用來收集數據，傳遞給Channel，可以處理各種類型、各種格式的日志數據,包括avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy、自定義。
- Sink:從Channel收集數據，運行在一個獨立線程,sink組件是用于把數據發送到目的地的組件，目的地包括hdfs、logger、avro、thrift、ipc、file、null、Hbase、solr、自定義。?
- Channel:連接 sources 和 sinks ，這個有點像一個隊列,source組件把數據收集來以后，臨時存放在channel中，即channel組件在agent中是專門用來存放臨時數據的——對采集到的數據進行簡單的緩存，可以存放在memory、jdbc、file等等。
- Events:可以是日志記錄、 avro 對象等。
Agent的概念
Flume以agent為最小的獨立運行單位。一個agent就是一個JVM,agent本身是一個Java進程，運行在日志收集節點—所謂日志收集節點就是服務器節點。?
單agent由Source、Sink和Channel三大組件構成，類似生產者、倉庫、消費者的架構.如下圖：
Event的概念?
flume的核心是把數據從數據源(source)收集過來，在將收集到的數據送到指定的目的地(sink)。為了保證輸送的過程一定成功，在送到目的地(sink)之前，會先緩存數據(channel),待數據真正到達目的地(sink)后，flume在刪除自己緩存的數據。?
在整個數據的傳輸的過程中，流動的是event，即事務保證是在event級別進行的。那么什么是event呢？—–event將傳輸的數據進行封裝，是flume傳輸數據的基本單位，如果是文本文件，通常是一行記錄，event也是事務的基本單位。event從source，流向channel，再到sink，本身為一個字節數組，并可攜帶headers(頭信息)信息。event代表著一個數據的最小完整單元，從外部數據源來，向外部的目的地去。?
為了方便大家理解，給出一張event的數據流向圖：?
一個完整的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的運行機制?
flume的核心就是一個agent，這個agent對外有兩個進行交互的地方，一個是接受數據的輸入——source，一個是數據的輸出sink，sink負責將數據發送到外部指定的目的地。source接收到數據之后，將數據發送給channel，chanel作為一個數據緩沖區會臨時存放這些數據，隨后sink會將channel中的數據發送到指定的地方—-例如HDFS等，注意：只有在sink將channel中的數據成功發送出去之后，channel才會將臨時數據進行刪除，這種機制保證了數據傳輸的可靠性與安全性。?
flume的廣義用法?
flume可以支持多級flume的agent，即flume可以前后相繼，例如sink可以將數據寫到下一個agent的source中，這樣的話就可以連成串了，可以整體處理了。flume還支持扇入(fan-in)、扇出(fan-out)。所謂扇入就是source可以接受多個輸入，所謂扇出就是sink可以將數據輸出多個目的地destination中。?
值得注意的是，Flume提供了大量內置的Source、Channel和Sink類型。不同類型的Source,Channel和Sink可以自由組合。組合方式基于用戶設置的配置文件，非常靈活。比如：Channel可以把事件暫存在內存里，也可以持久化到本地硬盤上。Sink可以把日志寫入HDFS, HBase，甚至是另外一個Source等等。Flume支持用戶建立多級流，也就是說，多個agent可以協同工作，并且支持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變量設置
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)驗證是否安裝成功
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$?
出現上面的信息，表示安裝成功了.
三、flume的案例
對于flume的原理其實很容易理解，我們更應該掌握flume的具體使用方法，flume提供了大量內置的Source、Channel和Sink類型。而且不同類型的Source、Channel和Sink可以自由組合—–組合方式基于用戶設置的配置文件，非常靈活。比如：Channel可以把事件暫存在內存里，也可以持久化到本地硬盤上。Sink可以把日志寫入HDFS, HBase，甚至是另外一個Source等等。下面我將用具體的案例詳述flume的具體用法。?
其實flume的用法很簡單—-書寫一個配置文件，在配置文件當中描述source、channel與sink的具體實現，而后運行一個agent實例，在運行agent實例的過程中會讀取配置文件的內容，這樣flume就會采集到數據。?
配置文件的編寫原則：
0)、案例1：Netcat?
NetCat Source：監聽一個指定的網絡端口，即只要應用程序向這個端口里面寫數據，這個source組件就可以獲取到信息。
1>從整體上描述代理agent中sources、sinks、channels所涉及到的組件??
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
2>詳細描述agent中每一個source、sink與channel的具體實現：即在描述source的時候，需要指定source到底是什么類型的，即這個source是接受文件的、還是接受http的、還是接受thrift的；對于sink也是同理，需要指定結果是輸出到HDFS中，還是Hbase中啊等等；對于channel需要指定是內存啊，還是數據庫啊，還是文件啊等等。
# 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連接起來
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
啟動agent的shell操作：
flume-ng agent -n a1 -c conf -f conf/example.file -Dflume.root.logger=DEBUG,console
參數說明：
-n 指定agent名稱(與配置文件中代理的名字相同)?
-c 指定flume中配置文件的目錄?
-f 指定配置文件?
-Dflume.root.logger=DEBUG,console 設置日志等級
1)、案例1：Avro
Avro可以發送一個給定的文件給Flume，Avro 源使用AVRO RPC機制。
a)創建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)啟動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)創建指定文件
mbp:apache-flume-1.7.0-bin$ echo "hello word" > log.00
d)使用avro-client發送文件
mbp:apache-flume-1.7.0-bin$ bin/flume-ng avro-client -c conf -H mbp -p 4141 -F log.00
e)在mbp的控制臺，可以看到以下信息，注意最后一行：
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
監聽一個指定的目錄，即只要應用程序向這個指定的目錄中添加新的文件，source組件就可以獲取到該信息，并解析該文件的內容，然后寫入到channle。寫入完成后，標記該文件已完成或者刪除該文件。
Spool監測配置的目錄下新增的文件，并將文件中的數據讀取出來。需要注意兩點：
1) 拷貝到spool目錄下的文件不可以再打開編輯。
2) spool目錄下不可包含相應的子目錄
a)創建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)啟動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的控制臺，可以看到以下相關信息：
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
監聽一個指定的命令，獲取一條命令的結果作為它的數據源?
常用的是tail -F file指令，即只要應用程序向日志(文件)里面寫數據，source組件就可以獲取到日志(文件)中最新的內容 。
EXEC執行一個給定的命令獲得輸出的源,如果要使用tail命令，必選使得file足夠大才能看到輸出內容
a)創建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)啟動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)生成足夠多的內容在文件里
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的控制臺，可以看到以下信息：
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監聽TCP的端口做為數據源?
a)創建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)啟動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)測試產生syslog
mbp:apache-flume-1.7.0-bin$ echo "hello idoall.org syslog" | nc localhost 5140
d)在mbp的控制臺，可以看到以下信息：
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)創建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)啟動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的控制臺，可以看到以下信息：
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 }?
總結Exec source：Exec source和Spooling Directory Source是兩種常用的日志采集的方式，其中Exec source可以實現對日志的實時采集，Spooling Directory Source在對日志的實時采集上稍有欠缺，盡管Exec source可以實現對日志的實時采集，但是當Flume不運行或者指令執行出錯時，Exec source將無法收集到日志數據，日志會出現丟失，從而無法保證收集日志的完整性。
6)、案例6：Avro Source
監聽一個指定的Avro 端口，通過Avro 端口可以獲取到Avro client發送過來的文件 。即只要應用程序通過Avro 端口發送文件，source組件就可以獲取到該文件中的內容。 其中 Sink：hdfs Channel：file?
(注：Avro和Thrift都是一些序列化的網絡端口–通過這些網絡端口可以接受或者發送信息，Avro可以發送一個給定的文件給Flume，Avro 源使用AVRO RPC機制)?
Avro Source運行原理如下圖：
flume配置文件的書寫是相當靈活的—-不同類型的Source、Channel和Sink可以自由組合！
最后對上面用的幾個flume source進行適當總結：?
① NetCat Source：監聽一個指定的網絡端口，即只要應用程序向這個端口里面寫數據，這個source組件?
就可以獲取到信息。?
②Spooling Directory Source：監聽一個指定的目錄，即只要應用程序向這個指定的目錄中添加新的文?
件，source組件就可以獲取到該信息，并解析該文件的內容，然后寫入到channle。寫入完成后，標記?
該文件已完成或者刪除該文件。?
③Exec Source：監聽一個指定的命令，獲取一條命令的結果作為它的數據源?
常用的是tail -F file指令，即只要應用程序向日志(文件)里面寫數據，source組件就可以獲取到日志(文件)中最新的內容 。?
④Avro Source：監聽一個指定的Avro 端口，通過Avro 端口可以獲取到Avro client發送過來的文件 。即只要應用程序通過Avro 端口發送文件，source組件就可以獲取到該文件中的內容。
7)、案例7：Hadoop sink
a)創建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)啟動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)測試產生syslog
mbp:apache-flume-1.7.0-bin$ echo "hello idoall flume -> hadoop testing one" | nc localhost 5140
d)在mbp的控制臺，可以看到以下信息：
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上再打開一個窗口，去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)創建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)啟動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)測試產生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下是否生成文件,默認每30秒生成一個新文件
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流從一個源到多個通道。有兩種模式的Fan out，分別是復制和復用。在復制的情況下，流的事件被發送到所有的配置通道。在復用的情況下，事件被發送到可用的渠道中的一個子集。Fan out流需要指定源和Fan out通道的規則。
這次我們需要用到mbp1,mbp2兩臺機器
a)在mbp1創建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創建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個配置文件復制到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個窗口，在mbp1和mbp2上同時啟動兩個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的任意一臺機器上，測試產生syslog
mbp1:apache-flume-1.7.0-bin$ echo "hello idoall.org syslog" | nc localhost 5140
f)在mbp1和mbp2的sink窗口，分別可以看到以下信息,這說明信息得到了同步：
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創建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
#映射允許每個值通道可以重疊。默認值可以包含任意數量的通道。
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創建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個配置文件復制到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個窗口，在mbp1和mbp2上同時啟動兩個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的任意一臺機器上，測試產生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 }
可以看到，根據header中不同的條件分布到不同的channel上
11)、案例11：Flume Sink Processors
failover的機器是一直發送給其中一個sink，當這個sink不可用的時候，自動發送到下一個sink。
a)在m1創建Flume_Sink_Processors配置文件
b)在m1創建Flume_Sink_Processors_avro配置文件
c)將2個配置文件復制到m2上一份
d)打開4個窗口，在m1和m2上同時啟動兩個flume agent
e)然后在m1或m2的任意一臺機器上，測試產生log
f)因為m2的優先級高，所以在m2的sink窗口，可以看到以下信息，而m1沒有：
g)這時我們停止掉m2機器上的sink(ctrl+c)，再次輸出測試數據：
h)可以在m1的sink窗口，看到讀取到了剛才發送的兩條測試數據：
i)我們再在m2的sink窗口中，啟動sink：
j)輸入兩批測試數據：
k)在m2的sink窗口，我們可以看到以下信息，因為優先級的關系，log消息會再次落到m2上：
?
12)、案例12：Load balancing Sink Processor
load balance type和failover不同的地方是，load balance有兩個配置，一個是輪詢，一個是隨機。兩種情況下如果被選擇的sink不可用，就會自動嘗試發送到下一個可用的sink上面。
a)在m1創建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
#這個是配置Load balancing的關鍵，需要有一個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創建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個配置文件復制到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個窗口，在mbp1和mbp2上同時啟動兩個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的任意一臺機器上，測試產生log，一行一行輸入，輸入太快，容易落到一臺機器上
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 }
說明輪詢模式起到了作用。
13)、案例13：Hbase sink
a)在測試之前，請先參考《ubuntu12.04+hadoop2.2.0+zookeeper3.4.5+hbase0.96.2+hive0.13.1分布式環境部署》將hbase啟動
b)然后將以下文件復制到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中已經存在
d)在m1創建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)啟動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)測試產生syslog
mbp:apache-flume-1.7.0-bin$ echo "hello idoall.org from flume" | nc localhost 5140
g)這時登錄到hbase中，可以發現新數據已經插入
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
經過這么多flume的例子測試，如果你全部做完后，會發現flume的功能真的很強大，可以進行各種搭配來完成你想要的工作，俗話說師傅領進門，修行在個人，如何能夠結合你的產品業務，將flume更好的應用起來，快去動手實踐吧。
?
?
?
?
?
轉載于:https://www.cnblogs.com/netbloomy/p/6666683.html
總結
以上是生活随笔為你收集整理的2、Flume1.7.0入门：安装、部署、及flume的案例的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 安装spring-tool-suite插
- 下一篇： (转) Docker EE/Docker