您好,登錄后才能下訂單哦!
本篇內(nèi)容主要講解“ Flume日志采集框架的使用方法”,感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷,實用性強。下面就讓小編來帶大家學習“ Flume日志采集框架的使用方法”吧!
Flume日志采集框架 安裝和部署 Flume運行機制 采集靜態(tài)文件到hdfs 采集動態(tài)日志文件到hdfs 兩個agent級聯(lián)
在一個完整的離線大數(shù)據(jù)處理系統(tǒng)中,除了hdfs+mapreduce+hive組成分析系統(tǒng)的核心之外,還需要數(shù)據(jù)采集、結(jié)果數(shù)據(jù)導出、任務調(diào)度等不可或缺的輔助系統(tǒng),而這些輔助工具在hadoop生態(tài)體系中都有便捷的開源框架,如圖所示:
Flume
是一個分布式、可靠、和高可用的海量日志采集、聚合和傳輸?shù)南到y(tǒng)。Flume
可以采集文件,socket
數(shù)據(jù)包、文件、文件夾、kafka
等各種形式源數(shù)據(jù),又可以將采集到的數(shù)據(jù)(下沉sink
)輸出到HDFS
、hbase
、hive
、kafka
等眾多外部存儲系統(tǒng)中。
對于一般的采集需求,通過對flume的簡單配置即可實現(xiàn)。
Flume
針對特殊場景也具備良好的自定義擴展能力,因此,flume
可以適用于大部分的日常數(shù)據(jù)采集場景。
Flume
分布式系統(tǒng)中最核心的角色是agent
,flume
采集系統(tǒng)就是由一個個agent
所連接起來形成,每一個agent
相當于一個數(shù)據(jù)傳遞員,內(nèi)部有三個組件:
Source
:采集組件,用于跟數(shù)據(jù)源對接,以獲取數(shù)據(jù)
Sink
:下沉組件,用于往下一級agent
傳遞數(shù)據(jù)或者往最終存儲系統(tǒng)傳遞數(shù)據(jù)
Channel
:傳輸通道組件,用于從source
將數(shù)據(jù)傳遞到sink
單個agent采集數(shù)據(jù):
多級agent之間串聯(lián):
1 下載安裝包apache-flume-1.9.0-bin.tar.gz
解壓
2 在conf
文件夾下的flume-env.sh
添加JAVA_HOME
export JAVA_HOME=/usr/local/bigdata/java/jdk1.8.0_211
3 根據(jù)采集的需求,添加采集方案配置文件,文件名可以任意取
具體可以看后面的示例
4 啟動flume
測試環(huán)境下:
$ bin/flume/-ng agent -c conf/ -f ./dir-hdfs.conf -n agent1 -Dflume.root.logger=INFO,console
命令說明:
-c
:指定flume自帶的配置文件目錄,不用自己修改
-f
:指定自己的配置文件,這里問當前文件夾下的dir-hdfs.conf
-n
:指定自己配置文件中使用那個agent
,對應的配置文件中定義的名字。
-Dflume.root.logger
:把日志打印在控制臺,類型為INFO
,這個只用于測試,后面將打印到日志文件中
生產(chǎn)中,啟動flume,應該把flume啟動在后臺:
nohup bin/flume-ng agent -c ./conf -f ./dir-hdfs.conf -n agent1 1>/dev/null 2>&1 &
某服務器的某特定目錄下,會不斷產(chǎn)生新的文件,每當有新文件出現(xiàn),就需要把文件采集到HDFS中去
在安裝目錄下添加文件dir-hdfs.conf
,然后添加配置信息。
先獲取agent
,命名為agent1
,后面的配置都跟在agent1
后面,也可以改為其他值,如agt1
,同一個配置文件中可以有多個配置配置方案,啟動agent
的時候獲取對應的名字就可以。
根據(jù)需求,首先定義以下3大要素
即source
——監(jiān)控文件目錄 : spooldir
spooldir
有如下特性:
監(jiān)視一個目錄,只要目錄中出現(xiàn)新文件,就會采集文件中的內(nèi)容
采集完成的文件,會被agent自動添加一個后綴:COMPLETED
(可修改)
所監(jiān)視的目錄中不允許重復出現(xiàn)相同文件名的文件
即sink——HDFS
文件系統(tǒng) : hdfs sink
即channel
——可用file channel
也可以用內(nèi)存channel
#定義三大組件的名稱 agent1.sources = source1 agent1.sinks = sink1 agent1.channels = channel1 # 配置source組件 agent1.sources.source1.type = spooldir agent1.sources.source1.spoolDir = /root/log/ agent1.sources.source1.fileSuffix=.FINISHED #文件每行的長度,注意這里如果事情文件每行超過這個長度會自動切斷,會導致數(shù)據(jù)丟失 agent1.sources.source1.deserializer.maxLineLength=5120 # 配置sink組件 agent1.sinks.sink1.type = hdfs agent1.sinks.sink1.hdfs.path =hdfs://Master:9000/access_log/%y-%m-%d/%H-%M agent1.sinks.sink1.hdfs.filePrefix = app_log agent1.sinks.sink1.hdfs.fileSuffix = .log agent1.sinks.sink1.hdfs.batchSize= 100 agent1.sinks.sink1.hdfs.fileType = DataStream agent1.sinks.sink1.hdfs.writeFormat =Text # roll:滾動切換:控制寫文件的切換規(guī)則 ## 按文件體積(字節(jié))來切 agent1.sinks.sink1.hdfs.rollSize = 512000 ## 按event條數(shù)切 agent1.sinks.sink1.hdfs.rollCount = 1000000 ## 按時間間隔切換文件 agent1.sinks.sink1.hdfs.rollInterval = 60 ## 控制生成目錄的規(guī)則 agent1.sinks.sink1.hdfs.round = true agent1.sinks.sink1.hdfs.roundValue = 10 agent1.sinks.sink1.hdfs.roundUnit = minute agent1.sinks.sink1.hdfs.useLocalTimeStamp = true # channel組件配置 agent1.channels.channel1.type = memory ## event條數(shù) agent1.channels.channel1.capacity = 500000 ##flume事務控制所需要的緩存容量600條event agent1.channels.channel1.transactionCapacity = 600 # 綁定source、channel和sink之間的連接 agent1.sources.source1.channels = channel1 agent1.sinks.sink1.channel = channel1
Channel
參數(shù)解釋:
capacity
:默認該通道中最大的可以存儲的event
數(shù)量
trasactionCapacity
:每次最大可以從source
中拿到或者送到sink
中的event
數(shù)量
keep-alive
:event
添加到通道中或者移出的允許時間
$ bin/flume/-ng agent -c conf/ -f dir-hdfs.conf -n agent1 -Dflume.root.logger=INFO,console
比如業(yè)務系統(tǒng)使用log4j生成的日志,日志內(nèi)容不斷增加,需要把追加到日志文件中的數(shù)據(jù)實時采集到hdfs
配置文件名稱:tail-hdfs.conf
根據(jù)需求,首先定義以下3大要素:
采集源,即source
——監(jiān)控文件內(nèi)容更新 : exec
tail -F file
下沉目標,即sink——HDFS
文件系統(tǒng) : hdfs sink
Source
和sink
之間的傳遞通道——channel
,可用file channel
也可以用 內(nèi)存channel
配置文件內(nèi)容:
# Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = exec a1.sources.r1.command = tail -F /root/app_weichat_login.log # Describe the sink agent1.sinks.sink1.type = hdfs agent1.sinks.sink1.hdfs.path =hdfs://Master:9000/app_weichat_login_log/%y-%m-%d/%H-%M agent1.sinks.sink1.hdfs.filePrefix = weichat_log agent1.sinks.sink1.hdfs.fileSuffix = .dat agent1.sinks.sink1.hdfs.batchSize= 100 agent1.sinks.sink1.hdfs.fileType = DataStream agent1.sinks.sink1.hdfs.writeFormat =Text agent1.sinks.sink1.hdfs.rollSize = 100 agent1.sinks.sink1.hdfs.rollCount = 1000000 agent1.sinks.sink1.hdfs.rollInterval = 60 agent1.sinks.sink1.hdfs.round = true agent1.sinks.sink1.hdfs.roundValue = 1 agent1.sinks.sink1.hdfs.roundUnit = minute agent1.sinks.sink1.hdfs.useLocalTimeStamp = true # 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
啟動命令:
bin/flume-ng agent -c conf -f conf/tail-hdfs.conf -n a1
從tail命令獲取數(shù)據(jù)發(fā)送到avro端口 另一個節(jié)點可配置一個avro源來中繼數(shù)據(jù),發(fā)送外部存儲
# Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = exec a1.sources.r1.command = tail -F /root/log/access.log # Describe the sink a1.sinks.k1.type = avro a1.sinks.k1.hostname = hdp-05 a1.sinks.k1.port = 4141 a1.sinks.k1.batch-size = 2 # 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
從avro端口接收數(shù)據(jù),下沉到hdfs
采集配置文件,avro-hdfs.conf
# Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source ##source中的avro組件是一個接收者服務 a1.sources.r1.type = avro a1.sources.r1.bind = hdp-05 a1.sources.r1.port = 4141 # Describe the sink a1.sinks.k1.type = hdfs a1.sinks.k1.hdfs.path = /flume/taildata/%y-%m-%d/ a1.sinks.k1.hdfs.filePrefix = tail- a1.sinks.k1.hdfs.round = true a1.sinks.k1.hdfs.roundValue = 24 a1.sinks.k1.hdfs.roundUnit = hour a1.sinks.k1.hdfs.rollInterval = 0 a1.sinks.k1.hdfs.rollSize = 0 a1.sinks.k1.hdfs.rollCount = 50 a1.sinks.k1.hdfs.batchSize = 10 a1.sinks.k1.hdfs.useLocalTimeStamp = true #生成的文件類型,默認是Sequencefile,可用DataStream,則為普通文本 a1.sinks.k1.hdfs.fileType = DataStream # 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
到此,相信大家對“ Flume日志采集框架的使用方法”有了更深的了解,不妨來實際操作一番吧!這里是億速云網(wǎng)站,更多相關內(nèi)容可以進入相關頻道進行查詢,關注我們,繼續(xù)學習!
免責聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。