日誌採集框架Flume

前言

在一個完整的大數據處理系統中，除了hdfs+mapreduce+hive組成分析系統的核心以外，還須要數據採集、結果數據導出、任務調度等不可或缺的輔助系統，而這些輔助工具在hadoop生態體系中都有便捷的開源框架，如圖所示：

 

 

 

 

 

 

1. 日誌採集框架Flume

1.1 Flume介紹

1.1.1 概述

u  Flume是一個分佈式、可靠、和高可用的海量日誌採集、聚合和傳輸的系統。

u  Flume能夠採集文件，socket數據包等各類形式源數據，又能夠將採集到的數據輸出到HDFS、hbase、hive、kafka等衆多外部存儲系統中

u  通常的採集需求，經過對flume的簡單配置便可實現

u  Flume針對特殊場景也具有良好的自定義擴展能力，所以，flume能夠適用於大部分的平常數據採集場景

 

1.1.2 運行機制

一、  Flume分佈式系統中最核心的角色是agent，flume採集系統就是由一個個agent所鏈接起來造成

二、  每個agent至關於一個數據傳遞員[M1] ，內部有三個組件：

a)       Source：採集源，用於跟數據源對接，以獲取數據

b)       Sink：下沉地，採集數據的傳送目的，用於往下一級agent傳遞數據或者往最終存儲系統傳遞數據

c)        Channel：angent內部的數據傳輸通道，用於從source將數據傳遞到sink

 

 

 

 

 

1.1.4 Flume採集系統結構圖

 

1. 簡單結構

單個agent採集數據

 

 

2. 複雜結構

多級agent之間串聯

 

 

 

1.2 Flume實戰案例

1.2.1 Flume的安裝部署

一、Flume的安裝很是簡單，只須要解壓便可，固然，前提是已有hadoop環境

上傳安裝包到數據源所在節點上

而後解壓  tar -zxvf apache-flume-1.6.0-bin.tar.gz

而後進入flume的目錄，修改conf下的flume-env.sh，在裏面配置JAVA_HOME

 

二、根據數據採集的需求配置採集方案，描述在配置文件中(文件名可任意自定義)

三、指定採集方案配置文件，在相應的節點上啓動flume agent

 

先用一個最簡單的例子來測試一下程序環境是否正常

一、先在flume的conf目錄下新建一個文件

vi   netcat-logger.conf

# 定義這個agent中各組件的名字 a1.sources = r1 a1.sinks = k1 a1.channels = c1   # 描述和配置source組件：r1 a1.sources.r1.type = netcat a1.sources.r1.bind = localhost a1.sources.r1.port = 44444   # 描述和配置sink組件：k1 a1.sinks.k1.type = logger   # 描述和配置channel組件，此處使用是內存緩存的方式 a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100   # 描述和配置source  channel   sink之間的鏈接關係 a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1

 

二、啓動agent去採集數據

bin/flume-ng agent -c conf -f conf/netcat-logger.conf -n a1  -Dflume.root.logger=INFO,console

-c conf   指定flume自身的配置文件所在目錄

-f conf/netcat-logger.con  指定咱們所描述的採集方案

-n a1  指定咱們這個agent的名字

三、測試

先要往agent採集監聽的端口上發送數據，讓agent有數據可採

隨便在一個能跟agent節點聯網的機器上

telnet anget-hostname  port   （telnet localhost 44444）

 

 

 

 

1.2.2 採集案例

一、採集目錄到HDFS

採集需求：某服務器的某特定目錄下，會不斷產生新的文件，每當有新文件出現，就須要把文件採集到HDFS中去

根據需求，首先定義如下3大要素

l  採集源，即source——監控文件目錄 :  spooldir

l  下沉目標，即sink——HDFS文件系統  :  hdfs sink

l  source和sink之間的傳遞通道——channel，可用file channel 也能夠用內存channel

 

配置文件編寫：

#定義三大組件的名稱 agent1.sources = source1 agent1.sinks = sink1 agent1.channels = channel1   # 配置source組件 agent1.sources.source1.type = spooldir agent1.sources.source1.spoolDir = /home/hadoop/logs/ agent1.sources.source1.fileHeader = false   #配置攔截器 agent1.sources.source1.interceptors = i1 agent1.sources.source1.interceptors.i1.type = host agent1.sources.source1.interceptors.i1.hostHeader = hostname   # 配置sink組件 agent1.sinks.sink1.type = hdfs agent1.sinks.sink1.hdfs.path =hdfs://hdp-node-01:9000/weblog/flume-collection/%y-%m-%d/%H-%M agent1.sinks.sink1.hdfs.filePrefix = access_log agent1.sinks.sink1.hdfs.maxOpenFiles = 5000 agent1.sinks.sink1.hdfs.batchSize= 100 agent1.sinks.sink1.hdfs.fileType = DataStream agent1.sinks.sink1.hdfs.writeFormat =Text agent1.sinks.sink1.hdfs.rollSize = 102400 agent1.sinks.sink1.hdfs.rollCount = 1000000 agent1.sinks.sink1.hdfs.rollInterval = 60 #agent1.sinks.sink1.hdfs.round = true #agent1.sinks.sink1.hdfs.roundValue = 10 #agent1.sinks.sink1.hdfs.roundUnit = minute agent1.sinks.sink1.hdfs.useLocalTimeStamp = true # Use a channel which buffers events in memory agent1.channels.channel1.type = memory agent1.channels.channel1.keep-alive = 120 agent1.channels.channel1.capacity = 500000 agent1.channels.channel1.transactionCapacity = 600   # Bind the source and sink to the channel agent1.sources.source1.channels = channel1 agent1.sinks.sink1.channel = channel1

 

Channel參數解釋：

capacity：默認該通道中最大的能夠存儲的event數量

trasactionCapacity：每次最大能夠從source中拿到或者送到sink中的event數量

keep-alive：event添加到通道中或者移出的容許時間

 

二、採集文件到HDFS

採集需求：好比業務系統使用log4j生成的日誌，日誌內容不斷增長，須要把追加到日誌文件中的數據實時採集到hdfs

 

根據需求，首先定義如下3大要素

l  採集源，即source——監控文件內容更新 :  exec  ‘tail -F file’

l  下沉目標，即sink——HDFS文件系統  :  hdfs sink

l  Source和sink之間的傳遞通道——channel，可用file channel 也能夠用 內存channel

 

配置文件編寫：

agent1.sources = source1 agent1.sinks = sink1 agent1.channels = channel1   # Describe/configure tail -F source1 agent1.sources.source1.type = exec agent1.sources.source1.command = tail -F /home/hadoop/logs/access_log agent1.sources.source1.channels = channel1   #configure host for source agent1.sources.source1.interceptors = i1 agent1.sources.source1.interceptors.i1.type = host agent1.sources.source1.interceptors.i1.hostHeader = hostname   # Describe sink1 agent1.sinks.sink1.type = hdfs #a1.sinks.k1.channel = c1 agent1.sinks.sink1.hdfs.path =hdfs://hdp-node-01:9000/weblog/flume-collection/%y-%m-%d/%H-%M agent1.sinks.sink1.hdfs.filePrefix = access_log agent1.sinks.sink1.hdfs.maxOpenFiles = 5000 agent1.sinks.sink1.hdfs.batchSize= 100 agent1.sinks.sink1.hdfs.fileType = DataStream agent1.sinks.sink1.hdfs.writeFormat =Text agent1.sinks.sink1.hdfs.rollSize = 102400 agent1.sinks.sink1.hdfs.rollCount = 1000000 agent1.sinks.sink1.hdfs.rollInterval = 60 agent1.sinks.sink1.hdfs.round = true agent1.sinks.sink1.hdfs.roundValue = 10 agent1.sinks.sink1.hdfs.roundUnit = minute agent1.sinks.sink1.hdfs.useLocalTimeStamp = true   # Use a channel which buffers events in memory agent1.channels.channel1.type = memory agent1.channels.channel1.keep-alive = 120 agent1.channels.channel1.capacity = 500000 agent1.channels.channel1.transactionCapacity = 600   # Bind the source and sink to the channel agent1.sources.source1.channels = channel1 agent1.sinks.sink1.channel = channel1