一、為什麼要用到Flume

        在以前搭建HAWQ資料倉庫實驗環境時，我使用Sqoop抽取從MySQL資料庫增量抽取資料到HDFS，然後用HAWQ的外部表進行訪問。這種方式只需要很少量的配置即可完成資料抽取任務，但缺點同樣明顯，那就是實時性。Sqoop使用MapReduce讀寫資料，而MapReduce是為了批處理場景設計的，目標是大吞吐量，並不太關心低延時問題。就像實驗中所做的，每天定時增量抽取資料一次。
        Flume是一個海量日誌採集、聚合和傳輸的系統，支援在日誌系統中定製各類資料傳送方，用於收集資料。同時，Flume提供對資料進行簡單處理，並寫到各種資料接受方的能力。Flume以流方式處理資料，可作為代理持續執行。當新的資料可用時，Flume能夠立即獲取資料並輸出至目標，這樣就可以在很大程度上解決實時性問題。
        Flume是最初只是一個日誌收集器，但隨著flume-ng-sql-source外掛的出現，使得Flume從關係資料庫採集資料成為可能。下面簡單介紹Flume，並詳細說明如何配置Flume將MySQL表資料準實時抽取到HDFS。

二、Flume簡介

1. Flume的概念

        Flume是分散式的日誌收集系統，它將各個伺服器中的資料收集起來並送到指定的地方去，比如說送到HDFS，簡單來說flume就是收集日誌的，其架構如圖1所示。
圖1

2. Event的概念 

        在這裡有必要先介紹一下Flume中event的相關概念：Flume的核心是把資料從資料來源（source）收集過來，在將收集到的資料送到指定的目的地（sink）。為了保證輸送的過程一定成功，在送到目的地（sink）之前，會先快取資料（channel）,待資料真正到達目的地（sink）後，Flume再刪除自己快取的資料。 
       在整個資料的傳輸的過程中，流動的是event，即事務保證是在event級別進行的。那麼什麼是event呢？Event將傳輸的資料進行封裝，是Flume傳輸資料的基本單位，如果是文字檔案，通常是一行記錄。Event也是事務的基本單位。Event從source，流向channel，再到sink，本身為一個位元組陣列，並可攜帶headers（頭資訊）資訊。Event代表著一個數據的最小完整單元，從外部資料來源來，向外部的目的地去。

3. Flume架構介紹 

        Flume之所以這麼神奇，是源於它自身的一個設計，這個設計就是agent。Agent本身是一個Java程序，執行在日誌收集節點——所謂日誌收集節點就是伺服器節點。 Agent裡面包含3個核心的元件：source、channel和sink，類似生產者、倉庫、消費者的架構。 

• Source：source元件是專門用來收集資料的，可以處理各種型別、各種格式的日誌資料,包括avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy、自定義。 
• Channel：source元件把資料收集來以後，臨時存放在channel中，即channel元件在agent中是專門用來存放臨時資料的——對採集到的資料進行簡單的快取，可以存放在memory、jdbc、file等等。 
• Sink：sink元件是用於把資料傳送到目的地的元件，目的地包括hdfs、logger、avro、thrift、ipc、file、null、Hbase、solr、自定義。 

4. Flume的執行機制 

        Flume的核心就是一個agent，這個agent對外有兩個進行互動的地方，一個是接受資料輸入的source，一個是資料輸出的sink，sink負責將資料傳送到外部指定的目的地。source接收到資料之後，將資料傳送給channel，chanel作為一個數據緩衝區會臨時存放這些資料，隨後sink會將channel中的資料傳送到指定的地方，例如HDFS等。注意：只有在sink將channel中的資料成功傳送出去之後，channel才會將臨時資料進行刪除，這種機制保證了資料傳輸的可靠性與安全性。 

三、安裝Hadoop和Flume

        我的實驗在HDP 2.5.0上進行，HDP安裝中包含Flume，只要配置Flume服務即可。HDP的安裝步驟參見“HAWQ技術解析（二） —— 安裝部署”

四、配置與測試

1. 建立MySQL資料庫表

        建立測試表並新增資料。

use test;

create table  wlslog  
(id         int not null,
 time_stamp varchar(40),
 category   varchar(40),
 type       varchar(40),
 servername varchar(40),
 code       varchar(40),
 msg        varchar(40),
 primary key ( id )
);

insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(1,'apr-8-2014-7:06:16-pm-pdt','notice','weblogicserver','adminserver','bea-000365','server state changed to standby');
insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(2,'apr-8-2014-7:06:17-pm-pdt','notice','weblogicserver','adminserver','bea-000365','server state changed to starting');
insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(3,'apr-8-2014-7:06:18-pm-pdt','notice','weblogicserver','adminserver','bea-000365','server state changed to admin');
insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(4,'apr-8-2014-7:06:19-pm-pdt','notice','weblogicserver','adminserver','bea-000365','server state changed to resuming');
insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(5,'apr-8-2014-7:06:20-pm-pdt','notice','weblogicserver','adminserver','bea-000361','started weblogic adminserver');
insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(6,'apr-8-2014-7:06:21-pm-pdt','notice','weblogicserver','adminserver','bea-000365','server state changed to running');
insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(7,'apr-8-2014-7:06:22-pm-pdt','notice','weblogicserver','adminserver','bea-000360','server started in running mode');
commit;

2. 建立相關目錄與檔案

（1）建立本地狀態檔案

mkdir -p /var/lib/flume
cd /var/lib/flume
touch sql-source.status
chmod -R 777 /var/lib/flume

（2）建立HDFS目標目錄

hdfs dfs -mkdir -p /flume/mysql
hdfs dfs -chmod -R 777 /flume/mysql

3. 準備JAR包

        從http://book2s.com/java/jar/f/flume-ng-sql-source/download-flume-ng-sql-source-1.3.7.html下載flume-ng-sql-source-1.3.7.jar檔案，並複製到Flume庫目錄。

cp flume-ng-sql-source-1.3.7.jar /usr/hdp/current/flume-server/lib/

        將MySQL JDBC驅動JAR包也複製到Flume庫目錄。

cp mysql-connector-java-5.1.17.jar /usr/hdp/current/flume-server/lib/mysql-connector-java.jar

4. 建立HAWQ外部表

create external table ext_wlslog
(id         int,
 time_stamp varchar(40),
 category   varchar(40),
 type       varchar(40),
 servername varchar(40),
 code       varchar(40),
 msg        varchar(40)
) location ('pxf://mycluster/flume/mysql?profile=hdfstextmulti') format 'csv' (quote=e'"'); 

5. 配置Flume

        在Ambari -> Flume -> Configs -> flume.conf中配置如下屬性：

agent.channels.ch1.type = memory
agent.sources.sql-source.channels = ch1
agent.channels = ch1
agent.sinks = HDFS

agent.sources = sql-source
agent.sources.sql-source.type = org.keedio.flume.source.SQLSource

agent.sources.sql-source.connection.url = jdbc:mysql://172.16.1.127:3306/test
agent.sources.sql-source.user = root
agent.sources.sql-source.password = 123456
agent.sources.sql-source.table = wlslog
agent.sources.sql-source.columns.to.select = *

agent.sources.sql-source.incremental.column.name = id
agent.sources.sql-source.incremental.value = 0

agent.sources.sql-source.run.query.delay=5000

agent.sources.sql-source.status.file.path = /var/lib/flume
agent.sources.sql-source.status.file.name = sql-source.status

agent.sinks.HDFS.channel = ch1
agent.sinks.HDFS.type = hdfs
agent.sinks.HDFS.hdfs.path = hdfs://mycluster/flume/mysql
agent.sinks.HDFS.hdfs.fileType = DataStream
agent.sinks.HDFS.hdfs.writeFormat = Text
agent.sinks.HDFS.hdfs.rollSize = 268435456
agent.sinks.HDFS.hdfs.rollInterval = 0
agent.sinks.HDFS.hdfs.rollCount = 0

        Flume在flume.conf檔案中指定Source、Channel和Sink相關的配置，各屬性描述如表1所示。

屬性	描述
agent.channels.ch1.type	Agent的channel型別
agent.sources.sql-source.channels	Source對應的channel名稱
agent.channels	Channel名稱
agent.sinks	Sink名稱
agent.sources	Source名稱
agent.sources.sql-source.type	Source型別
agent.sources.sql-source.connection.url	資料庫URL
agent.sources.sql-source.user	資料庫使用者名稱
agent.sources.sql-source.password	資料庫密碼
agent.sources.sql-source.table	資料庫表名
agent.sources.sql-source.columns.to.select	查詢的列
agent.sources.sql-source.incremental.column.name	增量列名
agent.sources.sql-source.incremental.value	增量初始值
agent.sources.sql-source.run.query.delay	發起查詢的時間間隔，單位是毫秒
agent.sources.sql-source.status.file.path	狀態檔案路徑
agent.sources.sql-source.status.file.name	狀態檔名稱
agent.sinks.HDFS.channel	Sink對應的channel名稱
agent.sinks.HDFS.type	Sink型別
agent.sinks.HDFS.hdfs.path	Sink路徑
agent.sinks.HDFS.hdfs.fileType	流資料的檔案型別
agent.sinks.HDFS.hdfs.writeFormat	資料寫入格式
agent.sinks.HDFS.hdfs.rollSize	目標檔案輪轉大小，單位是位元組
agent.sinks.HDFS.hdfs.rollInterval	hdfs sink間隔多長將臨時檔案滾動成最終目標檔案，單位是秒；如果設定成0，則表示不根據時間來滾動檔案
agent.sinks.HDFS.hdfs.rollCount	當events資料達到該數量時候，將臨時檔案滾動成目標檔案；如果設定成0，則表示不根據events資料來滾動檔案

表1

6. 執行Flume代理

        儲存上一步的設定，然後重啟Flume服務，如圖2所示。
圖2
        重啟後，狀態檔案已經記錄了將最新的id值7，如圖3所示。
圖3
        檢視目標路徑，生成了一個臨時檔案，其中有7條記錄，如圖4所示。
圖4
        查詢HAWQ外部表，結果也有全部7條資料，如圖5所示。
圖5
        至此，初始資料抽取已經完成。

7. 測試準實時增量抽取

        在源表中新增id為8、9、10的三條記錄。

use test;
insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(8,'apr-8-2014-7:06:22-pm-pdt','notice','weblogicserver','adminserver','bea-000360','server started in running mode');
insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(9,'apr-8-2014-7:06:22-pm-pdt','notice','weblogicserver','adminserver','bea-000360','server started in running mode');
insert into wlslog(id,time_stamp,category,type,servername,code,msg) values(10,'apr-8-2014-7:06:22-pm-pdt','notice','weblogicserver','adminserver','bea-000360','server started in running mode');
commit;

        5秒之後查詢HAWQ外部表，從圖6可以看到，已經查詢出全部10條資料，準實時增量抽取成功。
圖6

五、方案優缺點

        利用Flume採集關係資料庫表資料最大的優點是配置簡單，不用程式設計。相比tungsten-replicator的複雜性，Flume只要在flume.conf檔案中配置source、channel及sink的相關屬性，已經沒什麼難度了。而與現在很火的canal比較，雖然不夠靈活，但畢竟一行程式碼也不用寫。再有該方案採用普通SQL輪詢的方式實現，具有通用性，適用於所有關係庫資料來源。
        這種方案的缺點與其優點一樣突出，主要體現在以下幾方面。

• 在源庫上執行了查詢，具有入侵性。
• 通過輪詢的方式實現增量，只能做到準實時，而且輪詢間隔越短，對源庫的影響越大。
• 只能識別新增資料，檢測不到刪除與更新。
• 要求源庫必須有用於表示增量的欄位。

        即便有諸多侷限，但用Flume抽取關係庫資料的方案還是有一定的價值，特別是在要求快速部署、簡化程式設計，又能滿足需求的應用場景，對傳統的Sqoop方式也不失為一種有效的補充。

參考：

Flume架構以及應用介紹
Streaming MySQL Database Table Data to HDFS with Flume
how to read data from oracle using FLUME to kafka broker
https://github.com/keedio/flume-ng-sql-source