FLUME的產生背景

• 對於關係型資料庫和HDFS,Hive，等的資料，我們可以使用sqoop將資料進行匯入匯出操作，但對於一些日誌資訊（源端）的定時收集，這種方式顯然不能給予滿足，這時有人會想到使用shell指令碼的定時作業排程將日誌收集出來，但是這種方式在處理大的資料和可靠性方面也顯現出很多缺點，再比如日誌資訊的儲存與壓縮格式，任務的監控，這些顯然也不能滿足。
• 基於以上，FLUME這個分散式，高可靠，高可用，已經非常成熟的日誌資訊收集框架隨之誕生，他的簡單，輕便，健壯等特性使得他在資料收集方面使用的較為廣泛。

哪些方面可以使用到FLUME

• FLUME自身提供了多種sources(源端型別），多種sinks(傳輸端型別）使得他可以處理很多型別端的資料，並傳輸到多種指定的輸出端，基於輸出端的不同，可以大致分為：1）傳輸到HDFS(離線批處理）；2)傳輸到kafka等訊息中介軟體，再由像sparkstreaming流式化處理系統進行處理

FLUME基本架構

Event概念

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

Agent

• 
• 一個Agent就是一個Flume程序，agent本身是一個java程序，執行在日誌收集節點—所謂日誌收集節點就是伺服器節點。
• 一個Agent有三大元件：Source,Channel,Sink.但是一個Agent可以由多個Source,多個Channel,多個Sink構成，一個Source可以對應多個Channel，但是一個Channel只能對應一個Sink。最簡單的Agent是由一個Source，一個Channel，一個Sink構成。
• Source :負責從源端採集資料，輸出到channel中，常用的Source有exec/Spooling Directory/Taildir Source/NetCat
• Channel ：負責快取Source端來的資料，常用的Channel有Memory/File
• Sink : 處理Channel而來的資料寫到目標端，常用的Sink有HDFS/Logger/Avro/Kafka

元件種類

• Sources
  

• Channels
  

• Sinks
  

使用

• Flume的使用就是寫一個Agent配置檔案，根據場景和需求指定不同得元件型別，通過Channel將Source端和Sink端繫結串聯起來，然後用flume -ng 啟動flume程序

• eg:

  1. vi hello.conf     
  2. # Name the components on this agent
  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1
  
  6. # Describe/configure the source
  7. a1.sources.r1.type = netcat
  8. a1.sources.r1.bind = 0.0.0.0
  9. a1.sources.r1.port = 44444
  
  10. # Describe the sink
  11. a1.sinks.k1.type = logger
  
  12. # Use a channel which buffers events in memory
  13. a1.channels.c1.type = memory
  
  
  14. # Bind the source and sink to the channel
  15. a1.sources.r1.channels = c1
  16. a1.sinks.k1.channel = c1
  

• 啟動：

  1. ./flume-ng agent --name a1 --conf $FLUME_HOME/conf --conf-file $FLUME_HOME/conf/hello.conf -Dflume.root.logger=INFO,console
  

FLUME部署

• 下載：下載CDH版本 apache-flume-1.6.0-cdh5.7.0-bin

• 配置環境變數：

  1. export FLUME_HOME=/opt/software/apache-flume-1.6.0-cdh5.7.0-bin
  2. export PATH=$FLUME_HOME/bin:$PATH
  

• 指定JAVA_HOME:

  1.  cp flume-env.sh.template flume-env.sh
  2.  export JAVA_HOME=/usr/java/jdk1.8.0_45
  

測試案例

使用Exec Source Sink 到HDFS

• Exec Source：監聽一個指定的命令，獲取一條命令的結果作為它的資料來源
  常用的是tail -F file指令，即只要應用程式向日志(檔案)裡面寫資料，source元件就可以獲取到日誌(檔案)中最新的內容 。

• 配置檔案：

  1. # Name the components on this agent
  2. a1.sources = r1
  3. a1.sinks = k1
  4. a1.channels = c1
  
  5. # Describe/configure the source
  6. a1.sources.r1.type = exec
  7. a1.sources.r1.command = tail -F /home/hadoop/data/data.log
  
  8. # Describe the sink
  9. a1.sinks.k1.type = hdfs
  10. a1.sinks.k1.hdfs.path = hdfs://hadoop:9000/flume
  11. a1.sinks.k1.hdfs.writeFormat = Text
  12. a1.sinks.k1.hdfs.fileType = DataStream
  13. a1.sinks.k1.hdfs.rollInterval = 10
  14. a1.sinks.k1.hdfs.rollSize = 0
  15. a1.sinks.k1.hdfs.rollCount = 0
  16. a1.sinks.k1.hdfs.filePrefix = %Y-%m-%d-%H-%M-%S
  17. a1.sinks.k1.hdfs.useLocalTimeStamp = true
  
  
  18. # Use a channel which buffers events in memory
  19. a1.channels.c1.type=memory
  20. a1.channels.c1.capacity=10000
  21. a1.channels.c1.transactionCapacity=1000
  
  22. # Bind the source and sink to the channel
  23. a1.sources.r1.channels = c1
  24. a1.sinks.k1.channel = c1
  

• 啟動：

  1. ./flume-ng agent --name a1 --conf $FLUME_HOME/conf --conf-file $FLUME_HOME/conf/test3.conf  -Dflume.root.logger=INFO,console
  
  2. echo hadoop >data.log
  

• 檢視結果：

  1. [[email protected] ~]$ hdfs dfs -text /flume/2018-04-21-20-13-38.1524370418338 
  hadoop
  

使用Avro進行資料傳輸

• 以上的flume的使用方式都是啟動的agent和sink端在一個節點之上，對於多節點的方式資料傳輸,flume給我們提供了Avro的Sink端和Avro的Source端，這樣在多個Agent對接就提供瞭解決方案，在生產中的實際部署也多是使用Avro的方式。

使用Avro-client對接Avro source

• A機器向B機器傳輸日誌
  A機器：avro-client
  B機器：avro-source ，channel（memory） ，sink(logger)

• 配置資訊

  1. B機器的agent檔案：
  2. a1.sources=r1
  3. a1.sinks=k1
  4. a1.channels=c1
  
  5. a1.sources.r1.type=avro
  6. a1.sources.r1.bind=0.0.0.0  
  7. a1.sources.r1.port=44444  
  
  8. a1.channels.c1.type=memory
  
  
  9. a1.sinks.k1.type=logger
  
  10. a1.sinks.k1.channel=c1
  11. a1.sources.r1.channels=c1
  

• 執行命令：

  1. B機器上：
  2. ./flume-ng agent \
  3. --name a1 \
  4. --conf $FLUME_HOME/conf \
  5. --conf-file $FLUME_HOME/conf/avro.conf \
  6. -Dflume.root.logger=INFO,console
  
  7. 在A機器上：
  8. ./flume-ng avro-client --host 0.0.0.0 --port 44444  --filename /home/hadoop/data/input.txt
  

  使用此種方式只能一次性傳輸檔案，傳輸結束，程序也就自動結束了。

使用Avro Source和Avro Sink

• 定義A機器的agent:

  1. a1.sources=r1
  2. a1.sinks=k1
  3. a1.channels=c1
  
  
  4. a1.sources.r1.type=exec
  5. a1.sources.r1.command=tail -F /home/hadoop/data/data.log 
  6. a1.channels.c1.type=memory
  
  7. a1.sinks.k1.type=avro
  8. a1.sinks.k1.bind=0.0.0.0    //與B機器相對應
  9. a1.sinks.k1.port=44444       //與B機器相對應
  
  10. a1.sinks.k1.channel=c1
  11. a1.sources.r1.channels=c1 
  

• 定義B機器的agent:

  1. b1.sources=r1
  2. b1.sinks=k1
  3. b1.channels=c1
  
  4. b1.sources.r1.type=avro
  5. a1.sources.r1.bind = 0.0.0.0   
  6. a1.sources.r1.port = 44444   
  
  7. b1.channels.c1.type=memory
  
  8. b1.sinks.k1.type=logger
  
  9. b1.sinks.k1.channel=c1
  10. b1.sources.r1.channels=c1
  

*啟動：

    1. b機器先執行：
    2. ./flume-ng agent \
    3. --name b1 \
    4. --conf $FLUME_HOME/conf \
    5. --conf-file $FLUME_HOME/conf/avro_source.conf \
    6. -Dflume.root.logger=INFO,console   

    7. 然後在A機器上執行：

    8. ./flume-ng agent \
    9. --name a1 \
    10. --conf $FLUME_HOME/conf \
    11. --conf-file $FLUME_HOME/conf/avro_sink.conf \
    12. -Dflume.root.logger=INFO,console 


    13. [[email protected] data]$ echo aaa > data.log 
    14. [[email protected] data]$ echo 112121 > data.log 
    15. A機器目標目錄下檔案內容會被列印在B機器的控制檯上