Flume1.5.0入門：安裝、部署、及flume的案例

時間  2019-11-10
標籤 flume1.5.0 flume 入門 安裝 部署 案例 欄目 日誌分析 简体版
原文   原文鏈接

Flume1.5.0入門：安裝、部署、及flume的案例 

問題導讀
1.什麼是flume
2.flume的官方網站在哪裏？
3.flume有哪些術語？
4.如何配置flume數據源碼？






　　1、什麼是Flume?
　　flume 做爲 cloudera 開發的實時日誌收集系統，受到了業界的承認與普遍應用。Flume 初始的發行版本目前被統稱爲 Flume OG（original generation），屬於 cloudera。但隨着 FLume 功能的擴展，Flume OG 代碼工程臃腫、核心組件設計不合理、核心配置不標準等缺點暴露出來，尤爲是在 Flume OG 的最後一個發行版本 0.94.0 中，日誌傳輸不穩定的現象尤其嚴重，爲了解決這些問題，2011 年 10 月 22 號，cloudera 完成了 Flume-728，對 Flume 進行了里程碑式的改動：重構核心組件、核心配置以及代碼架構，重構後的版本統稱爲 Flume NG（next generation）；改動的另外一緣由是將 Flume 歸入 apache 旗下，cloudera Flume 更名爲 Apache Flume。

        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。
  • Sink        從Channel收集數據，運行在一個獨立線程。
  • Channel        鏈接 sources 和 sinks ，這個有點像一個隊列。
  • Events        能夠是日誌記錄、 avro 對象等。


　　Flume以agent爲最小的獨立運行單位。一個agent就是一個JVM。單agent由Source、Sink和Channel三大組件構成，以下圖：

 


　　值得注意的是，Flume提供了大量內置的Source、Channel和Sink類型。不一樣類型的Source,Channel和Sink能夠自由組合。組合方式基於用戶設置的配置文件，很是靈活。好比：Channel能夠把事件暫存在內存裏，也能夠持久化到本地硬盤上。Sink能夠把日誌寫入HDFS, HBase，甚至是另一個Source等等。Flume支持用戶創建多級流，也就是說，多個agent能夠協同工做，而且支持Fan-in、Fan-out、Contextual Routing、Backup Routes，這也正是NB之處。以下圖所示:

 



　　2、flume的官方網站在哪裏？
　　http://flume.apache.org/

　　3、在哪裏下載？
　　http://www.apache.org/dyn/closer.cgi/flume/1.5.0/apache-flume-1.5.0-bin.tar.gz

　　4、如何安裝？
　　　　1)將下載的flume包，解壓到/home/hadoop目錄中，你就已經完成了50%：）簡單吧
　　　　2)修改 flume-env.sh 配置文件,主要是JAVA_HOME變量設置

  1. root@m1:/home/hadoop/flume-1.5.0-bin# cp conf/flume-env.sh.template conf/flume-env.sh
  2. root@m1:/home/hadoop/flume-1.5.0-bin# vi conf/flume-env.sh
  3. # Licensed to the Apache Software Foundation (ASF) under one
  4. # or more contributor license agreements.  See the NOTICE file
  5. # distributed with this work for additional information
  6. # regarding copyright ownership.  The ASF licenses this file
  7. # to you under the Apache License, Version 2.0 (the
  8. # "License"); you may not use this file except in compliance
  9. # with the License.  You may obtain a copy of the License at
  10. #
  11. #     http://www.apache.org/licenses/LICENSE-2.0
  12. #
  13. # Unless required by applicable law or agreed to in writing, software
  14. # distributed under the License is distributed on an "AS IS" BASIS,
  15. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16. # See the License for the specific language governing permissions and
  17. # limitations under the License.

  19. # If this file is placed at FLUME_CONF_DIR/flume-env.sh, it will be sourced
  20. # during Flume startup.

  22. # Enviroment variables can be set here.

  24. JAVA_HOME=/usr/lib/jvm/java-7-oracle

  26. # Give Flume more memory and pre-allocate, enable remote monitoring via JMX
  27. #JAVA_OPTS="-Xms100m -Xmx200m -Dcom.sun.management.jmxremote"

  29. # Note that the Flume conf directory is always included in the classpath.
  30. #FLUME_CLASSPATH=""
複製代碼

              3)驗證是否安裝成功

  1. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng version
  2. Flume 1.5.0
  3. Source code repository: https://git-wip-us.apache.org/repos/asf/flume.git
  4. Revision: 8633220df808c4cd0c13d1cf0320454a94f1ea97
  5. Compiled by hshreedharan on Wed May  7 14:49:18 PDT 2014
  6. From source with checksum a01fe726e4380ba0c9f7a7d222db961f
  7. root@m1:/home/hadoop#
複製代碼
　　　　出現上面的信息，表示安裝成功了


　　5、flume的案例
　　　　1)案例1：Avro
　　　　Avro能夠發送一個給定的文件給Flume，Avro 源使用AVRO RPC機制。
　　　　　　a)建立agent配置文件

  1. root@m1:/home/hadoop#vi /home/hadoop/flume-1.5.0-bin/conf/avro.conf

  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1

  7. # Describe/configure the source
  8. a1.sources.r1.type = avro
  9. a1.sources.r1.channels = c1
  10. a1.sources.r1.bind = 0.0.0.0
  11. a1.sources.r1.port = 4141

  13. # Describe the sink
  14. a1.sinks.k1.type = logger

  16. # Use a channel which buffers events in memory
  17. a1.channels.c1.type = memory
  18. a1.channels.c1.capacity = 1000
  19. a1.channels.c1.transactionCapacity = 100

  21. # Bind the source and sink to the channel
  22. a1.sources.r1.channels = c1
  23. a1.sinks.k1.channel = c1
複製代碼



　　　　　　b)啓動flume agent a1

  1. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/avro.conf -n a1 -Dflume.root.logger=INFO,console
複製代碼

　　　　　　c)建立指定文件

  1. root@m1:/home/hadoop# echo "hello world" > /home/hadoop/flume-1.5.0-bin/log.00
複製代碼

　　　　　　d)使用avro-client發送文件

  1. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng avro-client -c . -H m1 -p 4141 -F /home/hadoop/flume-1.5.0-bin/log.00
複製代碼

　　　　　　d)使用avro-client發送文件


  1. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng avro-client -c . -H m1 -p 4141 -F /
複製代碼

　　　　　　f)在m1的控制檯，能夠看到如下信息，注意最後一行：


  1. root@m1:/home/hadoop/flume-1.5.0-bin/conf# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/avro.conf -n a1 -Dflume.root.logger=INFO,console
  2. Info: Sourcing environment configuration script /home/hadoop/flume-1.5.0-bin/conf/flume-env.sh
  3. Info: Including Hadoop libraries found via (/home/hadoop/hadoop-2.2.0/bin/hadoop) for HDFS access
  4. Info: Excluding /home/hadoop/hadoop-2.2.0/share/hadoop/common/lib/slf4j-api-1.7.5.jar from classpath
  5. Info: Excluding /home/hadoop/hadoop-2.2.0/share/hadoop/common/lib/slf4j-log4j12-1.7.5.jar from classpath
  6. ...
  7. 2014-08-10 10:43:25,112 (New I/O  worker #1) [INFO - org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.handleUpstream(NettyServer.java:171)] [id: 0x92464c4f, /192.168.1.50:59850 :> /192.168.1.50:4141] UNBOUND
  8. 2014-08-10 10:43:25,112 (New I/O  worker #1) [INFO - org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.handleUpstream(NettyServer.java:171)] [id: 0x92464c4f, /192.168.1.50:59850 :> /192.168.1.50:4141] CLOSED
  9. 2014-08-10 10:43:25,112 (New I/O  worker #1) [INFO - org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.channelClosed(NettyServer.java:209)] Connection to /192.168.1.50:59850 disconnected.
  10. 2014-08-10 10:43:26,718 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 68 65 6C 6C 6F 20 77 6F 72 6C 64                hello world }
複製代碼


　　　　2)案例2：Spool
　　　　Spool監測配置的目錄下新增的文件，並將文件中的數據讀取出來。須要注意兩點：
　　　　1) 拷貝到spool目錄下的文件不能夠再打開編輯。
　　　　2) spool目錄下不可包含相應的子目錄


　　　　　　a)建立agent配置文件

  1. root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/spool.conf

  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1

  7. # Describe/configure the source
  8. a1.sources.r1.type = spooldir
  9. a1.sources.r1.channels = c1
  10. a1.sources.r1.spoolDir = /home/hadoop/flume-1.5.0-bin/logs
  11. a1.sources.r1.fileHeader = true

  13. # Describe the sink
  14. a1.sinks.k1.type = logger

  16. # Use a channel which buffers events in memory
  17. a1.channels.c1.type = memory
  18. a1.channels.c1.capacity = 1000
  19. a1.channels.c1.transactionCapacity = 100

  21. # Bind the source and sink to the channel
  22. a1.sources.r1.channels = c1
  23. a1.sinks.k1.channel = c1
複製代碼

　　　　　　b)啓動flume agent a1

  1. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/spool.conf -n a1 -Dflume.root.logger=INFO,console
複製代碼

　　　　　　c)追加文件到/home/hadoop/flume-1.5.0-bin/logs目錄

  1. root@m1:/home/hadoop# echo "spool test1" > /home/hadoop/flume-1.5.0-bin/logs/spool_text.log
複製代碼

　　　　　d)在m1的控制檯，能夠看到如下相關信息：

  1. 14/08/10 11:37:13 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.
  2. 14/08/10 11:37:13 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.
  3. 14/08/10 11:37:14 INFO avro.ReliableSpoolingFileEventReader: Preparing to move file /home/hadoop/flume-1.5.0-bin/logs/spool_text.log to /home/hadoop/flume-1.5.0-bin/logs/spool_text.log.COMPLETED
  4. 14/08/10 11:37:14 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.
  5. 14/08/10 11:37:14 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.
  6. 14/08/10 11:37:14 INFO sink.LoggerSink: Event: { headers:{file=/home/hadoop/flume-1.5.0-bin/logs/spool_text.log} body: 73 70 6F 6F 6C 20 74 65 73 74 31                spool test1 }
  7. 14/08/10 11:37:15 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.
  8. 14/08/10 11:37:15 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.
  9. 14/08/10 11:37:16 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.
  10. 14/08/10 11:37:16 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.
  11. 14/08/10 11:37:17 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.
複製代碼



　　　　3)案例3：Exec
　　　　EXEC執行一個給定的命令得到輸出的源,若是要使用tail命令，必選使得file足夠大才能看到輸出內容

　　　　　　a)建立agent配置文件

  1. root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/exec_tail.conf

  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1

  7. # Describe/configure the source
  8. a1.sources.r1.type = exec
  9. a1.sources.r1.channels = c1
  10. a1.sources.r1.command = tail -F /home/hadoop/flume-1.5.0-bin/log_exec_tail

  12. # Describe the sink
  13. a1.sinks.k1.type = logger

  15. # Use a channel which buffers events in memory
  16. a1.channels.c1.type = memory
  17. a1.channels.c1.capacity = 1000
  18. a1.channels.c1.transactionCapacity = 100

  20. # Bind the source and sink to the channel
  21. a1.sources.r1.channels = c1
  22. a1.sinks.k1.channel = c1
複製代碼

　　　　　　b)啓動flume agent a1

  1. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/exec_tail.conf -n a1 -Dflume.root.logger=INFO,console
複製代碼

　　　　　　c)生成足夠多的內容在文件裏

  1. root@m1:/home/hadoop# for i in {1..100};do echo "exec tail$i" >> /home/hadoop/flume-1.5.0-bin/log_
複製代碼

　　　　　　e)在m1的控制檯，能夠看到如下信息：

  1. 2014-08-10 10:59:25,513 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 20 74 65 73 74       exec tail test }
  2. 2014-08-10 10:59:34,535 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 20 74 65 73 74       exec tail test }
  3. 2014-08-10 11:01:40,557 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 31                   exec tail1 }
  4. 2014-08-10 11:01:41,180 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 32                   exec tail2 }
  5. 2014-08-10 11:01:41,180 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 33                   exec tail3 }
  6. 2014-08-10 11:01:41,181 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 34                   exec tail4 }
  7. 2014-08-10 11:01:41,181 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 35                   exec tail5 }
  8. 2014-08-10 11:01:41,181 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 36                   exec tail6 }
  9. ....
  10. ....
  11. ....
  12. 2014-08-10 11:01:51,550 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 39 36                exec tail96 }
  13. 2014-08-10 11:01:51,550 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 39 37                exec tail97 }
  14. 2014-08-10 11:01:51,551 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 39 38                exec tail98 }
  15. 2014-08-10 11:01:51,551 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 39 39                exec tail99 }
  16. 2014-08-10 11:01:51,551 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 31 30 30             exec tail100 }
複製代碼



　　　　4)案例4：Syslogtcp
　　　　Syslogtcp監聽TCP的端口作爲數據源

　　　　　　a)建立agent配置文件

  1. root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/syslog_tcp.conf

  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1

  7. # Describe/configure the source
  8. a1.sources.r1.type = syslogtcp
  9. a1.sources.r1.port = 5140
  10. a1.sources.r1.host = localhost
  11. a1.sources.r1.channels = c1

  13. # Describe the sink
  14. a1.sinks.k1.type = logger

  16. # Use a channel which buffers events in memory
  17. a1.channels.c1.type = memory
  18. a1.channels.c1.capacity = 1000
  19. a1.channels.c1.transactionCapacity = 100

  21. # Bind the source and sink to the channel
  22. a1.sources.r1.channels = c1
  23. a1.sinks.k1.channel = c1
複製代碼

　　　　　　b)啓動flume agent a1


  1. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/syslog_tcp.conf -n a1 -Dflume.root.logger=INFO,console
複製代碼

　　　　　　c)測試產生syslog

  1. root@m1:/home/hadoop# echo "hello idoall.org syslog" | nc localhost 5140
複製代碼

　　　　　　d)在m1的控制檯，能夠看到如下信息：

  1. 14/08/10 11:41:45 INFO node.PollingPropertiesFileConfigurationProvider: Reloading configuration file:/home/hadoop/flume-1.5.0-bin/conf/syslog_tcp.conf
  2. 14/08/10 11:41:45 INFO conf.FlumeConfiguration: Added sinks: k1 Agent: a1
  3. 14/08/10 11:41:45 INFO conf.FlumeConfiguration: Processing:k1
  4. 14/08/10 11:41:45 INFO conf.FlumeConfiguration: Processing:k1
  5. 14/08/10 11:41:45 INFO conf.FlumeConfiguration: Post-validation flume configuration contains configuration for agents: [a1]
  6. 14/08/10 11:41:45 INFO node.AbstractConfigurationProvider: Creating channels
  7. 14/08/10 11:41:45 INFO channel.DefaultChannelFactory: Creating instance of channel c1 type memory
  8. 14/08/10 11:41:45 INFO node.AbstractConfigurationProvider: Created channel c1
  9. 14/08/10 11:41:45 INFO source.DefaultSourceFactory: Creating instance of source r1, type syslogtcp
  10. 14/08/10 11:41:45 INFO sink.DefaultSinkFactory: Creating instance of sink: k1, type: logger
  11. 14/08/10 11:41:45 INFO node.AbstractConfigurationProvider: Channel c1 connected to [r1, k1]
  12. 14/08/10 11:41:45 INFO node.Application: Starting new configuration:{ sourceRunners:{r1=EventDrivenSourceRunner: { source:org.apache.flume.source.SyslogTcpSource{name:r1,state:IDLE} }} sinkRunners:{k1=SinkRunner: { policy:org.apache.flume.sink.DefaultSinkProcessor@6538b14 counterGroup:{ name:null counters:{} } }} channels:{c1=org.apache.flume.channel.MemoryChannel{name: c1}} }
  13. 14/08/10 11:41:45 INFO node.Application: Starting Channel c1
  14. 14/08/10 11:41:45 INFO instrumentation.MonitoredCounterGroup: Monitored counter group for type: CHANNEL, name: c1: Successfully registered new MBean.
  15. 14/08/10 11:41:45 INFO instrumentation.MonitoredCounterGroup: Component type: CHANNEL, name: c1 started
  16. 14/08/10 11:41:45 INFO node.Application: Starting Sink k1
  17. 14/08/10 11:41:45 INFO node.Application: Starting Source r1
  18. 14/08/10 11:41:45 INFO source.SyslogTcpSource: Syslog TCP Source starting...
  19. 14/08/10 11:42:15 WARN source.SyslogUtils: Event created from Invalid Syslog data.
  20. 14/08/10 11:42:15 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 }
複製代碼



　　　　5)案例5：JSONHandler
　　　　　　a)建立agent配置文件

  1. root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/post_json.conf

  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1

  7. # Describe/configure the source
  8. a1.sources.r1.type = org.apache.flume.source.http.HTTPSource
  9. a1.sources.r1.port = 8888
  10. a1.sources.r1.channels = c1

  12. # Describe the sink
  13. a1.sinks.k1.type = logger

  15. # Use a channel which buffers events in memory
  16. a1.channels.c1.type = memory
  17. a1.channels.c1.capacity = 1000
  18. a1.channels.c1.transactionCapacity = 100

  20. # Bind the source and sink to the channel
  21. a1.sources.r1.channels = c1
  22. a1.sinks.k1.channel = c1
複製代碼

　　　　　　b)啓動flume agent a1


  1. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/post_json.conf -n a1 -Dflume.root.logger=INFO,console
複製代碼

　　　　　　c)生成JSON 格式的POST request

  1. root@m1:/home/hadoop# curl -X POST -d '[{ "headers" :{"a" : "a1","b" : "b1"},"body" : "idoall.org_body"}]' http://localhost:8888
複製代碼

　　　　　　d)在m1的控制檯，能夠看到如下信息：

  1. 14/08/10 11:49:59 INFO node.Application: Starting Channel c1
  2. 14/08/10 11:49:59 INFO instrumentation.MonitoredCounterGroup: Monitored counter group for type: CHANNEL, name: c1: Successfully registered new MBean.
  3. 14/08/10 11:49:59 INFO instrumentation.MonitoredCounterGroup: Component type: CHANNEL, name: c1 started
  4. 14/08/10 11:49:59 INFO node.Application: Starting Sink k1
  5. 14/08/10 11:49:59 INFO node.Application: Starting Source r1
  6. 14/08/10 11:49:59 INFO mortbay.log: Logging to org.slf4j.impl.Log4jLoggerAdapter(org.mortbay.log) via org.mortbay.log.Slf4jLog
  7. 14/08/10 11:49:59 INFO mortbay.log: jetty-6.1.26
  8. 14/08/10 11:50:00 INFO mortbay.log: Started SelectChannelConnector@0.0.0.0:8888
  9. 14/08/10 11:50:00 INFO instrumentation.MonitoredCounterGroup: Monitored counter group for type: SOURCE, name: r1: Successfully registered new MBean.
  10. 14/08/10 11:50:00 INFO instrumentation.MonitoredCounterGroup: Component type: SOURCE, name: r1 started
  11. 14/08/10 12:14:32 INFO sink.LoggerSink: Event: { headers:{b=b1, a=a1} body: 69 64 6F 61 6C 6C 2E 6F 72 67 5F 62 6F 64 79    idoall.org_body }
複製代碼

　　　　6)案例6：Hadoop sink
　　　　　　a)建立agent配置文件

  1. root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/hdfs_sink.conf

  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1

  7. # Describe/configure the source
  8. a1.sources.r1.type = syslogtcp
  9. a1.sources.r1.port = 5140
  10. a1.sources.r1.host = localhost
  11. a1.sources.r1.channels = c1

  13. # Describe the sink
  14. a1.sinks.k1.type = hdfs
  15. a1.sinks.k1.channel = c1
  16. a1.sinks.k1.hdfs.path = hdfs://m1:9000/user/flume/syslogtcp
  17. a1.sinks.k1.hdfs.filePrefix = Syslog
  18. a1.sinks.k1.hdfs.round = true
  19. a1.sinks.k1.hdfs.roundValue = 10
  20. a1.sinks.k1.hdfs.roundUnit = minute

  22. # Use a channel which buffers events in memory
  23. a1.channels.c1.type = memory
  24. a1.channels.c1.capacity = 1000
  25. a1.channels.c1.transactionCapacity = 100

  27. # Bind the source and sink to the channel
  28. a1.sources.r1.channels = c1
  29. a1.sinks.k1.channel = c1
複製代碼

　　　　　　b)啓動flume agent a1

  1. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/hdfs_sink.conf -n a1 -Dflume.root.logger=INFO,console
複製代碼

　　　　　　c)測試產生syslog

  1. root@m1:/home/hadoop# echo "hello idoall flume -> hadoop testing one" | nc localhost 5140
複製代碼

　　　　　　d)在m1的控制檯，能夠看到如下信息：

  1. 14/08/10 12:20:39 INFO instrumentation.MonitoredCounterGroup: Monitored counter group for type: CHANNEL, name: c1: Successfully registered new MBean.
  2. 14/08/10 12:20:39 INFO instrumentation.MonitoredCounterGroup: Component type: CHANNEL, name: c1 started
  3. 14/08/10 12:20:39 INFO node.Application: Starting Sink k1
  4. 14/08/10 12:20:39 INFO node.Application: Starting Source r1
  5. 14/08/10 12:20:39 INFO instrumentation.MonitoredCounterGroup: Monitored counter group for type: SINK, name: k1: Successfully registered new MBean.
  6. 14/08/10 12:20:39 INFO instrumentation.MonitoredCounterGroup: Component type: SINK, name: k1 started
  7. 14/08/10 12:20:39 INFO source.SyslogTcpSource: Syslog TCP Source starting...
  8. 14/08/10 12:21:46 WARN source.SyslogUtils: Event created from Invalid Syslog data.
  9. 14/08/10 12:21:49 INFO hdfs.HDFSSequenceFile: writeFormat = Writable, UseRawLocalFileSystem = false
  10. 14/08/10 12:21:49 INFO hdfs.BucketWriter: Creating hdfs://m1:9000/user/flume/syslogtcp//Syslog.1407644509504.tmp
  11. 14/08/10 12:22:20 INFO hdfs.BucketWriter: Closing hdfs://m1:9000/user/flume/syslogtcp//Syslog.1407644509504.tmp
  12. 14/08/10 12:22:20 INFO hdfs.BucketWriter: Close tries incremented
  13. 14/08/10 12:22:20 INFO hdfs.BucketWriter: Renaming hdfs://m1:9000/user/flume/syslogtcp/Syslog.1407644509504.tmp to hdfs://m1:9000/user/flume/syslogtcp/Syslog.1407644509504
  14. 14/08/10 12:22:20 INFO hdfs.HDFSEventSink: Writer callback called.
複製代碼

　　　　　　e)在m1上再打開一個窗口，去hadoop上檢查文件是否生成

  1. root@m1:/home/hadoop# /home/hadoop/hadoop-2.2.0/bin/hadoop fs -ls /user/flume/syslogtcp
  2. Found 1 items
  3. -rw-r--r--   3 root supergroup        155 2014-08-10 12:22 /user/flume/syslogtcp/Syslog.1407644509504
  4. root@m1:/home/hadoop# /home/hadoop/hadoop-2.2.0/bin/hadoop fs -cat /user/flume/syslogtcp/Syslog.1407644509504
  5. SEQ!org.apache.hadoop.io.LongWritable"org.apache.hadoop.io.BytesWritable^ ;>Gv$hello idoall flume -> hadoop testing one
複製代碼



　　　　7)案例7：File Roll Sink
　　　　　　a)建立agent配置文件

  1. root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/file_roll.conf

  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1

  7. # Describe/configure the source
  8. a1.sources.r1.type = syslogtcp
  9. a1.sources.r1.port = 5555
  10. a1.sources.r1.host = localhost
  11. a1.sources.r1.channels = c1

  13. # Describe the sink
  14. a1.sinks.k1.type = file_roll
  15. a1.sinks.k1.sink.directory = /home/hadoop/flume-1.5.0-bin/logs

  17. # Use a channel which buffers events in memory
  18. a1.channels.c1.type = memory
  19. a1.channels.c1.capacity = 1000
  20. a1.channels.c1.transactionCapacity = 100

  22. # Bind the source and sink to the channel
  23. a1.sources.r1.channels = c1
  24. a1.sinks.k1.channel = c1
複製代碼

　　　　　　b)啓動flume agent a1

  1. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/file_roll.conf -n a1 -Dflume.root.logger=INFO,console
複製代碼

　　　　　　c)測試產生log

  1. root@m1:/home/hadoop# echo "hello idoall.org syslog" | nc localhost 5555
  2. root@m1:/home/hadoop# echo "hello idoall.org syslog 2" | nc localhost 5555
複製代碼

　　　　　　d)查看/home/hadoop/flume-1.5.0-bin/logs下是否生成文件,默認每30秒生成一個新文件

  1. root@m1:/home/hadoop# ll /home/hadoop/flume-1.5.0-bin/logs
  2. 總用量 272
  3. drwxr-xr-x 3 root root   4096 Aug 10 12:50 ./
  4. drwxr-xr-x 9 root root   4096 Aug 10 10:59 ../
  5. -rw-r--r-- 1 root root     50 Aug 10 12:49 1407646164782-1
  6. -rw-r--r-- 1 root root      0 Aug 10 12:49 1407646164782-2
  7. -rw-r--r-- 1 root root      0 Aug 10 12:50 1407646164782-3
  8. root@m1:/home/hadoop# cat /home/hadoop/flume-1.5.0-bin/logs/1407646164782-1 /home/hadoop/flume-1.5.0-bin/logs/1407646164782-2
  9. hello idoall.org syslog
  10. hello idoall.org syslog 2
複製代碼



　　　　8)案例8：Replicating Channel Selector
　　　　Flume支持Fan out流從一個源到多個通道。有兩種模式的Fan out，分別是複製和複用。在複製的狀況下，流的事件被髮送到全部的配置通道。在複用的狀況下，事件被髮送到可用的渠道中的一個子集。Fan out流須要指定源和Fan out通道的規則。

　　　　此次咱們須要用到m1,m2兩臺機器

　　　　　　a)在m1建立replicating_Channel_Selector配置文件
  1. root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/replicating_Channel_Selector.conf

  3. a1.sources = r1
  4. a1.sinks = k1 k2
  5. a1.channels = c1 c2

  7. # Describe/configure the source
  8. a1.sources.r1.type = syslogtcp
  9. a1.sources.r1.port = 5140
  10. a1.sources.r1.host = localhost
  11. a1.sources.r1.channels = c1 c2
  12. a1.sources.r1.selector.type = replicating

  14. # Describe the sink
  15. a1.sinks.k1.type = avro
  16. a1.sinks.k1.channel = c1
  17. a1.sinks.k1.hostname = m1
  18. a1.sinks.k1.port = 5555

  20. a1.sinks.k2.type = avro
  21. a1.sinks.k2.channel = c2
  22. a1.sinks.k2.hostname = m2
  23. a1.sinks.k2.port = 5555

  25. # Use a channel which buffers events in memory
  26. a1.channels.c1.type = memory
  27. a1.channels.c1.capacity = 1000
  28. a1.channels.c1.transactionCapacity = 100

  30. a1.channels.c2.type = memory
  31. a1.channels.c2.capacity = 1000
  32. a1.channels.c2.transactionCapacity = 100
複製代碼

　　　　　　b)在m1建立replicating_Channel_Selector_avro配置文件
  1. root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/replicating_Channel_Selector_avro.conf

  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1

  7. # Describe/configure the source
  8. a1.sources.r1.type = avro
  9. a1.sources.r1.channels = c1
  10. a1.sources.r1.bind = 0.0.0.0
  11. a1.sources.r1.port = 5555

  13. # Describe the sink
  14. a1.sinks.k1.type = logger

  16. # Use a channel which buffers events in memory
  17. a1.channels.c1.type = memory
  18. a1.channels.c1.capacity = 1000
  19. a1.channels.c1.transactionCapacity = 100

  21. # Bind the source and sink to the channel
  22. a1.sources.r1.channels = c1
  23. a1.sinks.k1.channel = c1
複製代碼

　　　　　　c)在m1上將2個配置文件複製到m2上一份
  1. root@m1:/home/hadoop/flume-1.5.0-bin# scp -r /home/hadoop/flume-1.5.0-bin/conf/replicating_Channel_Selector.conf root@m2:/home/hadoop/flume-1.5.0-bin/conf/replicating_Channel_Selector.conf
  2. root@m1:/home/hadoop/flume-1.5.0-bin# scp -r /home/hadoop/flume-1.5.0-bin/conf/replicating_Channel_Selector_avro.conf root@m2:/home/hadoop/flume-1.5.0-bin/conf/replicating_Channel_Selector_avro.conf
複製代碼
　　　　　　d)打開4個窗口，在m1和m2上同時啓動兩個flume agent
  1. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/replicating_Channel_Selector_avro.conf -n a1 -Dflume.root.logger=INFO,console
  2. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/replicating_Channel_Selector.conf -n a1 -Dflume.root.logger=INFO,console
複製代碼

　　　　　　e)而後在m1或m2的任意一臺機器上，測試產生syslog
  1. root@m1:/home/hadoop# echo "hello idoall.org syslog" | nc localhost 5140
複製代碼

　　　　　　f)在m1和m2的sink窗口，分別能夠看到如下信息,這說明信息獲得了同步：
  1. 14/08/10 14:08:18 INFO ipc.NettyServer: Connection to /192.168.1.51:46844 disconnected.
  2. 14/08/10 14:08:52 INFO ipc.NettyServer: [id: 0x90f8fe1f, /192.168.1.50:35873 => /192.168.1.50:5555] OPEN
  3. 14/08/10 14:08:52 INFO ipc.NettyServer: [id: 0x90f8fe1f, /192.168.1.50:35873 => /192.168.1.50:5555] BOUND: /192.168.1.50:5555
  4. 14/08/10 14:08:52 INFO ipc.NettyServer: [id: 0x90f8fe1f, /192.168.1.50:35873 => /192.168.1.50:5555] CONNECTED: /192.168.1.50:35873
  5. 14/08/10 14:08:59 INFO ipc.NettyServer: [id: 0xd6318635, /192.168.1.51:46858 => /192.168.1.50:5555] OPEN
  6. 14/08/10 14:08:59 INFO ipc.NettyServer: [id: 0xd6318635, /192.168.1.51:46858 => /192.168.1.50:5555] BOUND: /192.168.1.50:5555
  7. 14/08/10 14:08:59 INFO ipc.NettyServer: [id: 0xd6318635, /192.168.1.51:46858 => /192.168.1.50:5555] CONNECTED: /192.168.1.51:46858
  8. 14/08/10 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 }
複製代碼



　　　　9)案例9：Multiplexing Channel Selector
　　　　　　a)在m1建立Multiplexing_Channel_Selector配置文件

  1. root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/Multiplexing_Channel_Selector.conf

  3. a1.sources = r1
  4. a1.sinks = k1 k2
  5. a1.channels = c1 c2

  7. # Describe/configure the source
  8. a1.sources.r1.type = org.apache.flume.source.http.HTTPSource
  9. a1.sources.r1.port = 5140
  10. a1.sources.r1.channels = c1 c2
  11. a1.sources.r1.selector.type = multiplexing

  13. a1.sources.r1.selector.header = type
  14. #映射容許每一個值通道能夠重疊。默認值能夠包含任意數量的通道。
  15. a1.sources.r1.selector.mapping.baidu = c1
  16. a1.sources.r1.selector.mapping.ali = c2
  17. a1.sources.r1.selector.default = c1

  19. # Describe the sink
  20. a1.sinks.k1.type = avro
  21. a1.sinks.k1.channel = c1
  22. a1.sinks.k1.hostname = m1
  23. a1.sinks.k1.port = 5555

  25. a1.sinks.k2.type = avro
  26. a1.sinks.k2.channel = c2
  27. a1.sinks.k2.hostname = m2
  28. a1.sinks.k2.port = 5555

  30. # Use a channel which buffers events in memory
  31. a1.channels.c1.type = memory
  32. a1.channels.c1.capacity = 1000
  33. a1.channels.c1.transactionCapacity = 100

  35. a1.channels.c2.type = memory
  36. a1.channels.c2.capacity = 1000
  37. a1.channels.c2.transactionCapacity = 100
複製代碼



　　　　　　b)在m1建立Multiplexing_Channel_Selector_avro配置文件
  1. root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/Multiplexing_Channel_Selector_avro.conf

  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1

  7. # Describe/configure the source
  8. a1.sources.r1.type = avro
  9. a1.sources.r1.channels = c1
  10. a1.sources.r1.bind = 0.0.0.0
  11. a1.sources.r1.port = 5555

  13. # Describe the sink
  14. a1.sinks.k1.type = logger

  16. # Use a channel which buffers events in memory
  17. a1.channels.c1.type = memory
  18. a1.channels.c1.capacity = 1000
  19. a1.channels.c1.transactionCapacity = 100

  21. # Bind the source and sink to the channel
  22. a1.sources.r1.channels = c1
  23. a1.sinks.k1.channel = c1
複製代碼



　　　　　　c)將2個配置文件複製到m2上一份
  1. root@m1:/home/hadoop/flume-1.5.0-bin# scp -r /home/hadoop/flume-1.5.0-bin/conf/Multiplexing_Channel_Selector.conf  root@m2:/home/hadoop/flume-1.5.0-bin/conf/Multiplexing_Channel_Selector.conf
  2. root@m1:/home/hadoop/flume-1.5.0-bin# scp -r /home/hadoop/flume-1.5.0-bin/conf/Multiplexing_Channel_Selector_avro.conf root@m2:/home/hadoop/flume-1.5.0-bin/conf/Multiplexing_Channel_Selector_avro.conf
複製代碼



　　　　　　d)打開4個窗口，在m1和m2上同時啓動兩個flume agent
  1. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/Multiplexing_Channel_Selector_avro.conf -n a1 -Dflume.root.logger=INFO,console
  2. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/Multiplexing_Channel_Selector.conf -n a1 -Dflume.root.logger=INFO,console
複製代碼



　　　　　　e)而後在m1或m2的任意一臺機器上，測試產生syslog
  1. root@m1:/home/hadoop# 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)在m1的sink窗口，能夠看到如下信息：
  1. 14/08/10 14:32:21 INFO node.Application: Starting Sink k1
  2. 14/08/10 14:32:21 INFO node.Application: Starting Source r1
  3. 14/08/10 14:32:21 INFO source.AvroSource: Starting Avro source r1: { bindAddress: 0.0.0.0, port: 5555 }...
  4. 14/08/10 14:32:21 INFO instrumentation.MonitoredCounterGroup: Monitored counter group for type: SOURCE, name: r1: Successfully registered new MBean.
  5. 14/08/10 14:32:21 INFO instrumentation.MonitoredCounterGroup: Component type: SOURCE, name: r1 started
  6. 14/08/10 14:32:21 INFO source.AvroSource: Avro source r1 started.
  7. 14/08/10 14:32:36 INFO ipc.NettyServer: [id: 0xcf00eea6, /192.168.1.50:35916 => /192.168.1.50:5555] OPEN
  8. 14/08/10 14:32:36 INFO ipc.NettyServer: [id: 0xcf00eea6, /192.168.1.50:35916 => /192.168.1.50:5555] BOUND: /192.168.1.50:5555
  9. 14/08/10 14:32:36 INFO ipc.NettyServer: [id: 0xcf00eea6, /192.168.1.50:35916 => /192.168.1.50:5555] CONNECTED: /192.168.1.50:35916
  10. 14/08/10 14:32:44 INFO ipc.NettyServer: [id: 0x432f5468, /192.168.1.51:46945 => /192.168.1.50:5555] OPEN
  11. 14/08/10 14:32:44 INFO ipc.NettyServer: [id: 0x432f5468, /192.168.1.51:46945 => /192.168.1.50:5555] BOUND: /192.168.1.50:5555
  12. 14/08/10 14:32:44 INFO ipc.NettyServer: [id: 0x432f5468, /192.168.1.51:46945 => /192.168.1.50:5555] CONNECTED: /192.168.1.51:46945
  13. 14/08/10 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 }
  14. 14/08/10 14:34:57 INFO sink.LoggerSink: Event: { headers:{type=qq} body: 69 64 6F 61 6C 6C 5F 54 45 53 54 33             idoall_TEST3 }
複製代碼

　　　　　g)在m2的sink窗口，能夠看到如下信息：

  1. 14/08/10 14:32:27 INFO node.Application: Starting Sink k1
  2. 14/08/10 14:32:27 INFO node.Application: Starting Source r1
  3. 14/08/10 14:32:27 INFO source.AvroSource: Starting Avro source r1: { bindAddress: 0.0.0.0, port: 5555 }...
  4. 14/08/10 14:32:27 INFO instrumentation.MonitoredCounterGroup: Monitored counter group for type: SOURCE, name: r1: Successfully registered new MBean.
  5. 14/08/10 14:32:27 INFO instrumentation.MonitoredCounterGroup: Component type: SOURCE, name: r1 started
  6. 14/08/10 14:32:27 INFO source.AvroSource: Avro source r1 started.
  7. 14/08/10 14:32:36 INFO ipc.NettyServer: [id: 0x7c2f0aec, /192.168.1.50:38104 => /192.168.1.51:5555] OPEN
  8. 14/08/10 14:32:36 INFO ipc.NettyServer: [id: 0x7c2f0aec, /192.168.1.50:38104 => /192.168.1.51:5555] BOUND: /192.168.1.51:5555
  9. 14/08/10 14:32:36 INFO ipc.NettyServer: [id: 0x7c2f0aec, /192.168.1.50:38104 => /192.168.1.51:5555] CONNECTED: /192.168.1.50:38104
  10. 14/08/10 14:32:44 INFO ipc.NettyServer: [id: 0x3d36f553, /192.168.1.51:48599 => /192.168.1.51:5555] OPEN
  11. 14/08/10 14:32:44 INFO ipc.NettyServer: [id: 0x3d36f553, /192.168.1.51:48599 => /192.168.1.51:5555] BOUND: /192.168.1.51:5555
  12. 14/08/10 14:32:44 INFO ipc.NettyServer: [id: 0x3d36f553, /192.168.1.51:48599 => /192.168.1.51:5555] CONNECTED: /192.168.1.51:48599
  13. 14/08/10 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上

　　　　10)案例10：Flume Sink Processors
　　　　failover的機器是一直髮送給其中一個sink，當這個sink不可用的時候，自動發送到下一個sink。

　　　　　　a)在m1建立Flume_Sink_Processors配置文件

  1. root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/Flume_Sink_Processors.conf

  3. a1.sources = r1
  4. a1.sinks = k1 k2
  5. a1.channels = c1 c2

  7. #這個是配置failover的關鍵，須要有一個sink group
  8. a1.sinkgroups = g1
  9. a1.sinkgroups.g1.sinks = k1 k2
  10. #處理的類型是failover
  11. a1.sinkgroups.g1.processor.type = failover
  12. #優先級，數字越大優先級越高，每一個sink的優先級必須不相同
  13. a1.sinkgroups.g1.processor.priority.k1 = 5
  14. a1.sinkgroups.g1.processor.priority.k2 = 10
  15. #設置爲10秒，固然能夠根據你的實際情況更改爲更快或者很慢
  16. a1.sinkgroups.g1.processor.maxpenalty = 10000

  18. # Describe/configure the source
  19. a1.sources.r1.type = syslogtcp
  20. a1.sources.r1.port = 5140
  21. a1.sources.r1.channels = c1 c2
  22. a1.sources.r1.selector.type = replicating


  25. # Describe the sink
  26. a1.sinks.k1.type = avro
  27. a1.sinks.k1.channel = c1
  28. a1.sinks.k1.hostname = m1
  29. a1.sinks.k1.port = 5555

  31. a1.sinks.k2.type = avro
  32. a1.sinks.k2.channel = c2
  33. a1.sinks.k2.hostname = m2
  34. a1.sinks.k2.port = 5555

  36. # Use a channel which buffers events in memory
  37. a1.channels.c1.type = memory
  38. a1.channels.c1.capacity = 1000
  39. a1.channels.c1.transactionCapacity = 100

  41. a1.channels.c2.type = memory
  42. a1.channels.c2.capacity = 1000
  43. a1.channels.c2.transactionCapacity = 100
複製代碼

　　　　　　b)在m1建立Flume_Sink_Processors_avro配置文件
  1. root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/Flume_Sink_Processors_avro.conf

  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1

  7. # Describe/configure the source
  8. a1.sources.r1.type = avro
  9. a1.sources.r1.channels = c1
  10. a1.sources.r1.bind = 0.0.0.0
  11. a1.sources.r1.port = 5555

  13. # Describe the sink
  14. a1.sinks.k1.type = logger

  16. # Use a channel which buffers events in memory
  17. a1.channels.c1.type = memory
  18. a1.channels.c1.capacity = 1000
  19. a1.channels.c1.transactionCapacity = 100

  21. # Bind the source and sink to the channel
  22. a1.sources.r1.channels = c1
  23. a1.sinks.k1.channel = c1
複製代碼

　　　　　　c)將2個配置文件複製到m2上一份
  1. root@m1:/home/hadoop/flume-1.5.0-bin# scp -r /home/hadoop/flume-1.5.0-bin/conf/Flume_Sink_Processors.conf  root@m2:/home/hadoop/flume-1.5.0-bin/conf/Flume_Sink_Processors.conf
  2. root@m1:/home/hadoop/flume-1.5.0-bin# scp -r /home/hadoop/flume-1.5.0-bin/conf/Flume_Sink_Processors_avro.conf root@m2:/home/hadoop/flume-1.5.0-bin/conf/Flume_Sink_Processors_avro.conf
複製代碼

　　　　　　d)打開4個窗口，在m1和m2上同時啓動兩個flume agent
  1. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/Flume_Sink_Processors_avro.conf -n a1 -Dflume.root.logger=INFO,console
  2. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/Flume_Sink_Processors.conf -n a1 -Dflume.root.logger=INFO,console
複製代碼

　　　　　　e)而後在m1或m2的任意一臺機器上，測試產生log
  1. root@m1:/home/hadoop# echo "idoall.org test1 failover" | nc localhost 5140
複製代碼

　　　　　　f)由於m2的優先級高，因此在m2的sink窗口，能夠看到如下信息，而m1沒有：
  1. 14/08/10 15:02:46 INFO ipc.NettyServer: Connection to /192.168.1.51:48692 disconnected.
  2. 14/08/10 15:03:12 INFO ipc.NettyServer: [id: 0x09a14036, /192.168.1.51:48704 => /192.168.1.51:5555] OPEN
  3. 14/08/10 15:03:12 INFO ipc.NettyServer: [id: 0x09a14036, /192.168.1.51:48704 => /192.168.1.51:5555] BOUND: /192.168.1.51:5555
  4. 14/08/10 15:03:12 INFO ipc.NettyServer: [id: 0x09a14036, /192.168.1.51:48704 => /192.168.1.51:5555] CONNECTED: /192.168.1.51:48704
  5. 14/08/10 15:03:26 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 }
複製代碼

　　　　　　g)這時咱們中止掉m2機器上的sink(ctrl+c)，再次輸出測試數據：
  1. root@m1:/home/hadoop# echo "idoall.org test2 failover" | nc localhost 5140
複製代碼

　　　　　　h)能夠在m1的sink窗口，看到讀取到了剛纔發送的兩條測試數據：
  1. 14/08/10 15:02:46 INFO ipc.NettyServer: Connection to /192.168.1.51:47036 disconnected.
  2. 14/08/10 15:03:12 INFO ipc.NettyServer: [id: 0xbcf79851, /192.168.1.51:47048 => /192.168.1.50:5555] OPEN
  3. 14/08/10 15:03:12 INFO ipc.NettyServer: [id: 0xbcf79851, /192.168.1.51:47048 => /192.168.1.50:5555] BOUND: /192.168.1.50:5555
  4. 14/08/10 15:03:12 INFO ipc.NettyServer: [id: 0xbcf79851, /192.168.1.51:47048 => /192.168.1.50:5555] CONNECTED: /192.168.1.51:47048
  5. 14/08/10 15:07:56 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 }
  6. 14/08/10 15:07:56 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 }
複製代碼

　　　　　　i)咱們再在m2的sink窗口中，啓動sink：
  1. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/Flume_Sink_Processors_avro.conf -n a1 -Dflume.root.logger=INFO,console
複製代碼

　　　　　　j)輸入兩批測試數據：
  1. root@m1:/home/hadoop# echo "idoall.org test3 failover" | nc localhost 5140 && echo "idoall.org test4 failover" | nc localhost 5140
複製代碼

　　　　　k)在m2的sink窗口，咱們能夠看到如下信息，由於優先級的關係，log消息會再次落到m2上：
  1. 14/08/10 15:09:47 INFO node.Application: Starting Sink k1
  2. 14/08/10 15:09:47 INFO node.Application: Starting Source r1
  3. 14/08/10 15:09:47 INFO source.AvroSource: Starting Avro source r1: { bindAddress: 0.0.0.0, port: 5555 }...
  4. 14/08/10 15:09:47 INFO instrumentation.MonitoredCounterGroup: Monitored counter group for type: SOURCE, name: r1: Successfully registered new MBean.
  5. 14/08/10 15:09:47 INFO instrumentation.MonitoredCounterGroup: Component type: SOURCE, name: r1 started
  6. 14/08/10 15:09:47 INFO source.AvroSource: Avro source r1 started.
  7. 14/08/10 15:09:54 INFO ipc.NettyServer: [id: 0x96615732, /192.168.1.51:48741 => /192.168.1.51:5555] OPEN
  8. 14/08/10 15:09:54 INFO ipc.NettyServer: [id: 0x96615732, /192.168.1.51:48741 => /192.168.1.51:5555] BOUND: /192.168.1.51:5555
  9. 14/08/10 15:09:54 INFO ipc.NettyServer: [id: 0x96615732, /192.168.1.51:48741 => /192.168.1.51:5555] CONNECTED: /192.168.1.51:48741
  10. 14/08/10 15:09:57 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 }
  11. 14/08/10 15:10:43 INFO ipc.NettyServer: [id: 0x12621f9a, /192.168.1.50:38166 => /192.168.1.51:5555] OPEN
  12. 14/08/10 15:10:43 INFO ipc.NettyServer: [id: 0x12621f9a, /192.168.1.50:38166 => /192.168.1.51:5555] BOUND: /192.168.1.51:5555
  13. 14/08/10 15:10:43 INFO ipc.NettyServer: [id: 0x12621f9a, /192.168.1.50:38166 => /192.168.1.51:5555] CONNECTED: /192.168.1.50:38166
  14. 14/08/10 15:10:43 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 }
  15. 14/08/10 15:10:43 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 }
複製代碼



　　　　11)案例11：Load balancing Sink Processor
　　　　load balance type和failover不一樣的地方是，load balance有兩個配置，一個是輪詢，一個是隨機。兩種狀況下若是被選擇的sink不可用，就會自動嘗試發送到下一個可用的sink上面。

　　　　　　a)在m1建立Load_balancing_Sink_Processors配置文件

  1. root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/Load_balancing_Sink_Processors.conf

  3. a1.sources = r1
  4. a1.sinks = k1 k2
  5. a1.channels = c1

  7. #這個是配置Load balancing的關鍵，須要有一個sink group
  8. a1.sinkgroups = g1
  9. a1.sinkgroups.g1.sinks = k1 k2
  10. a1.sinkgroups.g1.processor.type = load_balance
  11. a1.sinkgroups.g1.processor.backoff = true
  12. a1.sinkgroups.g1.processor.selector = round_robin

  14. # Describe/configure the source
  15. a1.sources.r1.type = syslogtcp
  16. a1.sources.r1.port = 5140
  17. a1.sources.r1.channels = c1


  20. # Describe the sink
  21. a1.sinks.k1.type = avro
  22. a1.sinks.k1.channel = c1
  23. a1.sinks.k1.hostname = m1
  24. a1.sinks.k1.port = 5555

  26. a1.sinks.k2.type = avro
  27. a1.sinks.k2.channel = c1
  28. a1.sinks.k2.hostname = m2
  29. a1.sinks.k2.port = 5555

  31. # Use a channel which buffers events in memory
  32. a1.channels.c1.type = memory
  33. a1.channels.c1.capacity = 1000
  34. a1.channels.c1.transactionCapacity = 100
複製代碼



　　　　　　b)在m1建立Load_balancing_Sink_Processors_avro配置文件
  1. root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/Load_balancing_Sink_Processors_avro.conf

  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1

  7. # Describe/configure the source
  8. a1.sources.r1.type = avro
  9. a1.sources.r1.channels = c1
  10. a1.sources.r1.bind = 0.0.0.0
  11. a1.sources.r1.port = 5555

  13. # Describe the sink
  14. a1.sinks.k1.type = logger

  16. # Use a channel which buffers events in memory
  17. a1.channels.c1.type = memory
  18. a1.channels.c1.capacity = 1000
  19. a1.channels.c1.transactionCapacity = 100

  21. # Bind the source and sink to the channel
  22. a1.sources.r1.channels = c1
  23. a1.sinks.k1.channel = c1
複製代碼

　　　　　　c)將2個配置文件複製到m2上一份
  1. root@m1:/home/hadoop/flume-1.5.0-bin# scp -r /home/hadoop/flume-1.5.0-bin/conf/Load_balancing_Sink_Processors.conf  root@m2:/home/hadoop/flume-1.5.0-bin/conf/Load_balancing_Sink_Processors.conf
  2. root@m1:/home/hadoop/flume-1.5.0-bin# scp -r /home/hadoop/flume-1.5.0-bin/conf/Load_balancing_Sink_Processors_avro.conf root@m2:/home/hadoop/flume-1.5.0-bin/conf/Load_balancing_Sink_Processors_avro.conf
複製代碼

　　　　　　d)打開4個窗口，在m1和m2上同時啓動兩個flume agent

  1. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/Load_balancing_Sink_Processors_avro.conf -n a1 -Dflume.root.logger=INFO,console
  2. root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/Load_balancing_Sink_Processors.conf -n a1 -Dflume.root.logger=INFO,console
複製代碼

　　　　　　e)而後在m1或m2的任意一臺機器上，測試產生log，一行一行輸入，輸入太快，容易落到一臺機器上
  1. root@m1:/home/hadoop# echo "idoall.org test1" | nc localhost 5140
  2. root@m1:/home/hadoop# echo "idoall.org test2" | nc localhost 5140
  3. root@m1:/home/hadoop# echo "idoall.org test3" | nc localhost 5140
  4. root@m1:/home/hadoop# echo "idoall.org test4" | nc localhost 5140
複製代碼

　　　　　　f)在m1的sink窗口，能夠看到如下信息：
  1. 14/08/10 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 }
  2. 14/08/10 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)在m2的sink窗口，能夠看到如下信息：
  1. 14/08/10 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 }
  2. 14/08/10 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 }
複製代碼



　　　　說明輪詢模式起到了做用。

　　　　12)案例12：Hbase sink


　　　　　　a)在測試以前，請先參考《ubuntu12.04+hadoop2.2.0+zookeeper3.4.5+hbase0.96.2+hive0.13.1分佈式環境部署》將hbase啓動

　　　　　　b)而後將如下文件複製到flume中：

  1. cp /home/hadoop/hbase-0.96.2-hadoop2/lib/protobuf-java-2.5.0.jar /home/hadoop/flume-1.5.0-bin/lib
  2. cp /home/hadoop/hbase-0.96.2-hadoop2/lib/hbase-client-0.96.2-hadoop2.jar /home/hadoop/flume-1.5.0-bin/lib
  3. cp /home/hadoop/hbase-0.96.2-hadoop2/lib/hbase-common-0.96.2-hadoop2.jar /home/hadoop/flume-1.5.0-bin/lib
  4. cp /home/hadoop/hbase-0.96.2-hadoop2/lib/hbase-protocol-0.96.2-hadoop2.jar /home/hadoop/flume-1.5.0-bin/lib
  5. cp /home/hadoop/hbase-0.96.2-hadoop2/lib/hbase-server-0.96.2-hadoop2.jar /home/hadoop/flume-1.5.0-bin/lib
  6. cp /home/hadoop/hbase-0.96.2-hadoop2/lib/hbase-hadoop2-compat-0.96.2-hadoop2.jar /home/hadoop/flume-1.5.0-bin/lib
  7. cp /home/hadoop/hbase-0.96.2-hadoop2/lib/hbase-hadoop-compat-0.96.2-hadoop2.jar /home/hadoop/flume-1.5.0-bin/lib@@@
  8. cp /home/hadoop/hbase-0.96.2-hadoop2/lib/htrace-core-2.04.jar /home/hadoop/flume-1.5.0-bin/lib
複製代碼

　　　　　　c)確保test_idoall_org表在hbase中已經存在

　　　　　　d)在m1建立hbase_simple配置文件
  1. root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/hbase_simple.conf

  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1

  7. # Describe/configure the source
  8. a1.sources.r1.type = syslogtcp
  9. a1.sources.r1.port = 5140
  10. a1.sources.r1.host = localhost
  11. a1.sources.r1.channels = c1

  13. # Describe the sink
  14. a1.sinks.k1.type = logger
  15. a1.sinks.k1.type = hbase
  16. a1.sinks.k1.table = test_idoall_org
  17. a1.sinks.k1.columnFamily = name
  18. a1.sinks.k1.column = idoall
  19. a1.sinks.k1.serializer =  org.apache.flume.sink.hbase.RegexHbaseEventSerializer
  20. a1.sinks.k1.channel = memoryChannel

  22. # Use a channel which buffers events in memory
  23. a1.channels.c1.type = memory
  24. a1.channels.c1.capacity = 1000
  25. a1.channels.c1.transactionCapacity = 100

  27. # Bind the source and sink to the channel
  28. a1.sources.r1.channels = c1
  29. a1.sinks.k1.channel = c1
複製代碼

　　　　　　e)啓動flume agent
  1. /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/hbase_simple.conf -n a1 -Dflume.root.logger=INFO,console
複製代碼

　　　　　　f)測試產生syslog
  1. root@m1:/home/hadoop# echo "hello idoall.org from flume" | nc localhost 5140
複製代碼

　　　　　　g)這時登陸到hbase中，能夠發現新數據已經插入
  1. root@m1:/home/hadoop# /home/hadoop/hbase-0.96.2-hadoop2/bin/hbase shell
  2. 2014-08-10 16:09:48,984 INFO  [main] Configuration.deprecation: hadoop.native.lib is deprecated. Instead, use io.native.lib.available
  3. HBase Shell; enter 'help<RETURN>' for list of supported commands.
  4. Type "exit<RETURN>" to leave the HBase Shell
  5. Version 0.96.2-hadoop2, r1581096, Mon Mar 24 16:03:18 PDT 2014

  7. hbase(main):001:0> list
  8. TABLE                                                                                                                                                                                                                  
  9. SLF4J: Class path contains multiple SLF4J bindings.
  10. SLF4J: Found binding in [jar:file:/home/hadoop/hbase-0.96.2-hadoop2/lib/slf4j-log4j12-1.6.4.jar!/org/slf4j/impl/StaticLoggerBinder.class]
  11. SLF4J: Found binding in [jar:file:/home/hadoop/hadoop-2.2.0/share/hadoop/common/lib/slf4j-log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]
  12. SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
  13. hbase2hive_idoall                                                                                                                                                                                                      
  14. hive2hbase_idoall                                                                                                                                                                                                      
  15. test_idoall_org                                                                                                                                                                                                        
  16. 3 row(s) in 2.6880 seconds

  18. => ["hbase2hive_idoall", "hive2hbase_idoall", "test_idoall_org"]
  19. hbase(main):002:0> scan "test_idoall_org"
  20. ROW                                                    COLUMN+CELL                                                                                                                                                     
  21. 10086                                                 column=name:idoall, timestamp=1406424831473, value=idoallvalue                                                                                                  
  22. 1 row(s) in 0.0550 seconds

  24. hbase(main):003:0> scan "test_idoall_org"
  25. ROW                                                    COLUMN+CELL                                                                                                                                                     
  26. 10086                                                 column=name:idoall, timestamp=1406424831473, value=idoallvalue                                                                                                  
  27. 1407658495588-XbQCOZrKK8-0                            column=name:payload, timestamp=1407658498203, value=hello idoall.org from flume                                                                                 
  28. 2 row(s) in 0.0200 seconds

  30. hbase(main):004:0> quit
複製代碼
通過這麼多flume的例子測試，若是你所有作完後，會發現flume的功能真的很強大，能夠進行各類搭配來完成你想要的工做，俗話說師傅領進門，修行在我的，如何可以結合你的產品業務，將flume更好的應用起來，快去動手實踐吧。
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程