[TOC] # 分析 采集需求：某服務器的某特定目錄下，會不斷產生新的文件，每當有新文件出現，就需要把文件采集到HDFS中去 根據需求，首先定義以下3大要素 * 采集源，即source——監控文件目錄 : spooldir * 下沉目標，即sink——HDFS文件系統 : hdfs sink * source和sink之間的傳遞通道——channel，可用file channel 也可以用內存memory channel # 配置文件 ~~~ #定義三大組件的名稱 agent1.sources = source1 agent1.sinks = sink1 agent1.channels = channel1 # 配置source組件 agent1.sources.source1.type = spooldir # 對那個目錄進行監聽上傳 agent1.sources.source1.spoolDir = /root/data/ # 上傳成功的文件結尾加上擴展名 agent1.sources.source1.fileSuffix=.COMPLETED # 是否使用當前文件頭,無論如何加存儲的絕對路徑的文件名 agent1.sources.source1.fileHeader = true # 忽略所有以.tmp結尾的文件,不上傳 agent1.sources.source1.ignorePattern=([^ ]*\.tmp) #配置攔截器 agent1.sources.source1.interceptors = i1 agent1.sources.source1.interceptors.i1.type = timestamp # 配置sink組件 # 類型是hdfs agent1.sinks.sink1.type = hdfs # 在hdfs上產生的目錄 agent1.sinks.sink1.hdfs.path =/weblog/flume-collection/%y-%m-%d/%H-%M # 文件的前綴,在hdfs上的前綴 agent1.sinks.sink1.hdfs.filePrefix = access_log # 最大文件打開數量 agent1.sinks.sink1.hdfs.maxOpenFiles = 5000 # 批次大小,就是文件達到多少條才提交到hdfs agent1.sinks.sink1.hdfs.batchSize = 100 # 當前文件存儲數據類型,還可以用壓縮格式 # 文件格式：當前是SequenceFile，DataStream或CompressedStream（1）DataStream不會壓縮輸出文件，請不要設置codeC（2）CompressedStream需要使用可用的codeC設置hdfs.codeC agent1.sinks.sink1.hdfs.fileType = DataStream # 文件的格式類型 agent1.sinks.sink1.hdfs.writeFormat =Text # 最小冗余數,不設置為0的話,滾動策略就失效 agent1.sinks.sink1.hdfs.minBlockReplicas=0 # 達到下面的三個任何一個就按照那個標準生成一個新文件 #設置每個文件的滾動大小 agent1.sinks.sink1.hdfs.rollSize = 102400 #滾動生成的文件按行數生成,0表示和event無關 agent1.sinks.sink1.hdfs.rollCount = 100 #多久生成一個新文件,秒 agent1.sinks.sink1.hdfs.rollInterval = 10 # 整體就是每10分鐘滾動生成一個目錄 #開啟滾動生成目錄 agent1.sinks.sink1.hdfs.round = true #多長時間會創建一個新的文件夾 agent1.sinks.sink1.hdfs.roundValue = 10 #單位為分鐘 agent1.sinks.sink1.hdfs.roundUnit = minute #是否使用本地時間戳 agent1.sinks.sink1.hdfs.useLocalTimeStamp = true # 管道的類型 agent1.channels.channel1.type = memory # 管道的容量,字節 agent1.channels.channel1.capacity = 500000 # 事務的類型,多少條之后source推送到channel或者channel推送到sinks agent1.channels.channel1.transactionCapacity = 600 # 多久之后將數據從source移動到channel,channel移動到sink agent1.channels.channel1.keep-alive = 120 # Bind the source and sink to the channel # 對應的source,channel進行組裝 agent1.sources.source1.channels = channel1 agent1.sinks.sink1.channel = channel1 ~~~ # 測試 啟動 ~~~ flume-ng agent -c conf -f fhd.conf -n agent1 -Dflume.root.logger=INFO,console ~~~ fhd.conf換成你自己寫的,不同的目錄加上目錄 -n代表上面定義的agent的名字 啟動后可以看到打印的日志,只要`/root/data/`下面有文件就會移動到hdfs 采集完的文件后面會有這個后綴`.COMPLETED` **flume的source采用spooldir時！ 目錄下面不允許存放同名的文件，否則報錯！** # 注意 在使用Spooling Directory Source時 1. 不要在監控目錄中創建并持續修改文件 2. 上傳完成的文件會以.COMPLERED結尾(fileSuffix) 3. 被監控文件夾每500毫秒掃描一次文件變動(pollDelay屬性默認值) # 其他組件：Interceptor(攔截器) **用于Source的一組Interceptor**，按照預設的順序在必要地方裝飾和過濾events。 內建的Interceptors允許增加event的headers比如：時間戳、主機名、靜態標記等等 定制的interceptors可以通過內省event payload（讀取原始日志），實現自己的業務邏輯（很強大）