本教程的目的：

1. flume 跟 hadoop 有什麼關係
2. 介紹flume的結構
3. 介紹flume的安裝
4. 做一個簡單的收集網路傳送過來的文字(NetCat Source)並儲存到日誌檔案(Logger Sink)的例子

flume 跟 hadoop 有什麼關係

hadoop是一個分散式系統，跟hadoop配合的一般也是分散式系統，分散式系統帶來的就是分散式日誌，分散式日誌帶來1. 日誌數量多 2. 日誌資料量大， 所以無論是採集分散式的日誌還是儲存海量的日誌到hadoop，都需要一個日誌收集系統，這就是flume。不過其實關係也不是太大，日誌方面沒有太大需求的人其實可以跳過flume的學習

flume 結構

以下是一張解釋flume如何工作的圖


介紹一下概念
Event: 事件是從Flume agent發過來的一個數據流單元。事件從 Source 到 Channel 到 Sink。
Flume agen:  是一個jvm 程序。這個程序負責將外部的來源產生的Event轉發到外部的目的地
Source: 負責消費有特殊格式的Event。這些Event會通過一個外部的來源(比如一個web伺服器)傳遞給Source. 比如一個 AvroSource 用來接收由客戶端或者在同一個流程上其他的Flume agent 發過來的 Avro Event (注意：flume的agent是可以一個接一個的串聯起來的)
當Source接收到Event的時候，它會把Event儲存到Channel(通道)裡面。Channel被動的儲存和保持Event直到Sink消費掉它
打個比方，有一種Channel叫 FileChannel，它使用檔案系統來儲存檔案
Sink:用來從Channel中移除Event，並且把Event放到外部資源庫，比如hdfs(這種Sink叫HDFSEventSink)。或者也可以繼續把Event推送給在同一個流程中的下一個Source。

flume一般是這樣用的

或者是這樣的

安裝 flume

flume有三個元件
flume-ng — flume本身
flume-ng-agent — 把flume agent做成一個服務的指令碼
flume-ng-doc — Flume 文件

$ yum install -y flume-ng flume-ng-agent flume-ng-doc

測試一下

$ flume-ng help

會輸出幫助資訊

使用flume

flume監聽NetCat例子

Step1

介紹下 NetCat Source ： 這個source會監聽一個埠，然後把每行通過這個埠發過來的文本當做一個事件傳給channel
介紹下 Logger Sink: 採用日誌的形式來消費Event。日誌級別為INFO
介紹下 Memory Channel : 使用記憶體來暫存Event

$ vim /etc/flume-ng/conf/example.conf
 

貼上以下文字進example.conf並儲存

# example.conf: A single-node Flume configuration


# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1


# 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


# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

STEP 2 執行 flume

$ flume-ng agent --conf conf --conf-file /etc/flume-ng/conf/example.conf --name a1 -Dflume.root.logger=INFO,console

如果沒有錯誤的話，shell會保持監聽狀態，此時監聽44444埠

STEP 3 測試 flume

我們新開啟一個ssh連線，然後使用curl 給 44444 埠傳送一條測試文字："hello world"

$ curl -X GET http://localhost:44444?w=helloworld
OK
OK
OK
OK
OK

這時去看之前監聽的那個ssh視窗會打印出

14/07/04 11:39:12 INFO sink.LoggerSink: Event: { headers:{} body: 47 45 54 20 2F 3F 77 3D 68 65 6C 6C 6F 77 6F 72 GET /?w=hellowor }
14/07/04 11:39:12 INFO sink.LoggerSink: Event: { headers:{} body: 55 73 65 72 2D 41 67 65 6E 74 3A 20 63 75 72 6C User-Agent: curl }
14/07/04 11:39:12 INFO sink.LoggerSink: Event: { headers:{} body: 48 6F 73 74 3A 20 6C 6F 63 61 6C 68 6F 73 74 3A Host: localhost: }
14/07/04 11:39:12 INFO sink.LoggerSink: Event: { headers:{} body: 41 63 63 65 70 74 3A 20 2A 2F 2A 0D             Accept: */*. }
14/07/04 11:39:12 INFO sink.LoggerSink: Event: { headers:{} body: 0D    

以上是flume的最簡單的例子，接下來介紹一個稍微複雜一點，但是真正實用的例子

Flume 收集log4j日誌檔案並寫入hdfs的例子

在這個例子中我們由本機將日誌傳送到host1上，並由host1上的flume將日誌儲存到hdfs上 。架構圖
我們會用到 log4j的一個Appender FlumeAppender 這個外掛，是由Flume開發的適用於log4j的Appender

step1

修改 /etc/flume-ng/conf/flume.conf 的內容為

# source, channel, sink definition
agent.channels = mem-channel
agent.sources = log4j-avro-source
agent.sinks = hdfs-sink

# Channel
# Define a  memory channel called mem-channel on agent
agent.channels.mem-channel.type = memory

# Source
# Define an Avro source called log4j-avro-channel on agent and tell it
# to bind to host1:12345. Connect it to channel mem-channel.
agent.sources.log4j-avro-source.type = avro
agent.sources.log4j-avro-source.bind = 192.168.1.126
agent.sources.log4j-avro-source.port = 12345
agent.sources.log4j-avro-source.channels = mem-channel

# Sink 
# Define a logger sink that simply logs all events it receives
# and connect it to the other end of the same channel.
agent.sinks.hdfs-sink.type = hdfs
agent.sinks.hdfs-sink.hdfs.path = hdfs://mycluster/flume/events/
agent.sinks.hdfs-sink.channel = mem-channel

然後重啟flume

service flume-ng-agent restart

step2

在hdfs上建立flume要寫的資料夾

sudo -u flume hdfs dfs -mkdir -p /flume/events/

step3

建立一個簡單的maven專案叫 play-flume
pom.xml  的內容是

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>org.crazycake</groupId>
  <artifactId>play-flume</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>

  <name>play-flume</name>
  <url>http://maven.apache.org</url>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>

  <dependencies>
		<dependency>
  		<groupId>log4j</groupId>
  		<artifactId>log4j</artifactId>
  		<version>1.2.17</version>
  	</dependency>
	<dependency>
		<groupId>org.apache.flume.flume-ng-clients</groupId>
		<artifactId>flume-ng-log4jappender</artifactId>
		<version>1.4.0</version>
	</dependency>
		<dependency>
			<groupId>junit</groupId>
			<artifactId>junit</artifactId>
			<version>4.11</version>
		</dependency>
	</dependencies>
	<build>
		<finalName>${project.artifactId}-${timestamp}</finalName>
		<plugins>
			<!-- compiler -->
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-compiler-plugin</artifactId>
				<version>2.5.1</version>
				<inherited>true</inherited>
				<configuration>
					<source>1.6</source>
					<target>1.6</target>
				</configuration>
			</plugin>

			<!-- jar -->
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-jar-plugin</artifactId>
				<version>2.4</version>
				<extensions>false</extensions>
				<inherited>true</inherited>
			</plugin>
		</plugins>
	</build>
</project>

增加log4j.properties  到 src/main/resources 資料夾下，如果沒有這個資料夾就自己建一個，並新增到 java build path 作為一個source folder 。
log4j.properties 內容為

# Define the root logger with appender file
log4j.rootLogger = DEBUG, stdout, flume

# Redirect log messages to console
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target=System.out
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n
 
# Define the flume appender
log4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppender
log4j.appender.flume.Hostname = host1
log4j.appender.flume.Port = 12345
log4j.appender.flume.UnsafeMode = false
log4j.appender.flume.layout=org.apache.log4j.PatternLayout

建立一個 FlumeLog.java

package org.crazycake.play_flume;

import org.apache.log4j.Logger;

public class FlumeLog {
    public static void main(String[] args)
    {
        Logger logger = Logger.getLogger(App.class);
        logger.info("hello world");
        logger.info("My name is alex");
        logger.info("How are you?");
    }
}

step4

執行 FlumeLog.java ，然後等幾秒鐘就可以看到在 /flume/events 目錄下生成的日誌檔案，不過是序列化成LongWritable 後的日誌，無法直接檢視
你也可以用 HBaseSink 直接把日誌寫到Hbase裡面，這樣就更省事一點，再結合上 Phoenix 可以直接寫SQL日誌查出來，不過就喪失了 MapReduce 的特性，不過真的要做MapReduce也不是不可以，就自己寫程式把日誌匯出來，再跑Map Reduce 就行了，具體的選擇看應用場景決定，如果你只是查日誌，那用Hbase就行了

真實環境下的架構

這個例子只是一個最簡單的情況，真實環境的日誌情況下，會有多臺機器收集log4j生成的日誌並統一存放到hdfs上。比如我們從host1 和 host2 上都收集日誌，並通過其中一臺機器的flume將日誌儲存到hdfs上。架構圖 
這樣做的好處是

1. 分散風險，提高容錯率，因為flume本身有一個快取機制，如果寫入hdfs的flume出問題了，比如關機，或者升級，web伺服器的日誌還是可以收集，只是暫時先放在本機的硬碟上
2. 緩解hdfs的寫入壓力，由於日誌在各個層級被快取了，並以一定的速度寫入hdfs，這樣不會造成hdfs的機器負載過大以致影響正常業務

參考資料