大數(shù)據(jù)技術(shù)原理與操作應(yīng)用 第8章 Flume日志采集系統(tǒng)

第8章Flume日志采集系統(tǒng)·Flume的可靠性保證·Flume攔截器·案例——日志采集·Flume概述·Flume的基本使用

·Flume采集方案配置說明

?學(xué)習(xí)目標熟悉熟悉了解Flume的概念與作用12熟悉Flume的運行機制和可靠性保證掌握Flume的基本使用方法和安裝部署3了解熟悉熟悉掌握4熟悉案例——日志采集的編寫本章將針對Flume系統(tǒng)的基本概念、安裝及應(yīng)用方式詳細講解，讓讀者掌握Flume系統(tǒng)的原理與應(yīng)用。章節(jié)概要在大數(shù)據(jù)系統(tǒng)的開發(fā)中，數(shù)據(jù)收集工作無疑是開發(fā)者首要解決的一個難題，但由于生產(chǎn)數(shù)據(jù)的源頭豐富多樣，其中包含網(wǎng)站日志數(shù)據(jù)、后臺監(jiān)控數(shù)據(jù)、用戶瀏覽網(wǎng)頁數(shù)據(jù)等，數(shù)據(jù)工程師要想將它們分門別類的采集到HDFS系統(tǒng)中，就可以使用ApacheFlume（數(shù)據(jù)采集）系統(tǒng)。

8.1Flume介紹Flume概述

Flume原是Cloudera公司提供的一個高可用的、高可靠的、分布式海量日志采集、聚合和傳輸系統(tǒng)，而后納入到了Apache旗下，作為一個頂級開源項目。ApacheFlume不僅只限于日志數(shù)據(jù)的采集，由于Flume采集的數(shù)據(jù)源是可定制的，因此Flume還可用于傳輸大量事件數(shù)據(jù)，包括但不限于網(wǎng)絡(luò)流量數(shù)據(jù)、社交媒體生成的數(shù)據(jù)、電子郵件消息以及幾乎任何可能的數(shù)據(jù)源。

8.1Flume介紹Flume運行機制Flume的核心是把數(shù)據(jù)從數(shù)據(jù)源（例如Web服務(wù)器）通過數(shù)據(jù)采集器（Source）收集過來，再將收集的數(shù)據(jù)通過緩沖通道（Channel）匯集到指定的接收器（Sink）。

Flume以Agent為最小的獨立運行單位。一個Agent就是一個JVM。單Agent由Source、Sink和Channel三大組件構(gòu)成,

8.1Flume介紹FlumeSources在編寫Flume采集方案時，首先必須明確采集的數(shù)據(jù)源類型、出處；接著，根據(jù)這些信息與Flume已提供支持的FlumeSources進行匹配，選擇對應(yīng)的數(shù)據(jù)采集器類型（即sources.type）；再根據(jù)選擇的數(shù)據(jù)采集器類型，配置必要和非必要的數(shù)據(jù)采集器屬性，F(xiàn)lume提供并支持的FlumeSources種類如下所示。AvroSourceThriftSourceExecSourceJMSSourceSpoolingDirectorySourceTwitter1%firehoseSourceKafkaSourceNetCatTCPSourceNetCatUDPSourceSequenceGeneratorSourceSyslogTCPSourceMultiportSyslogTCPSourceSyslogUDPSourceHTTPSourceStressSourceAvroLegacySourceThriftLegacySourceCustomSourceScribeSourceTaildirSource

8.1Flume介紹FlumeSources

監(jiān)聽Avro端口并從外部Avro客戶端流中接收event數(shù)據(jù)，當(dāng)與另一個FlumeAgent上的AvroSink配對時，可創(chuàng)建分層集合拓撲，利用AvroSource可以實現(xiàn)多級流動、扇出流、扇入流等效果，AvroSource常用配置屬性如下。屬性名稱默認值相關(guān)說明channels–

type–組件類型名需必須是avrobind–要監(jiān)聽的主機名或IP地址port–要監(jiān)聽的服務(wù)端口threads–要生成的工作線程的最大數(shù)目sslfalse將此設(shè)置為true以啟用SSL加密，則還必須指定“keystore”和“keystore-password”keystore–SSL所必需的通往Java秘鑰存儲路徑keystore-password–SSL所必需的Java密鑰存儲的密碼

8.1Flume介紹FlumeSourcesSpoolingDirectorySource允許對指定磁盤上的文件目錄進行監(jiān)控來提取數(shù)據(jù)，它將查看文件的指定目錄的新增文件，并將文件中的數(shù)據(jù)讀取出來。SpoolingDirectorySource常用配置屬性如下表所示。屬性名稱默認值相關(guān)說明channels–

type–組件類型名需必須是spooldirspoolDir–從中讀取文件的目錄fileSuffix.COMPLETED附加到完全攝取的文件后綴deletePolicynever何時刪除已完成的文件：never或immediatefileHeaderfalse是否添加存儲絕對路徑文件名的標頭includePattern^.*$正則表達式，指定要包含的文件ignorePattern^$正則表達式指定要忽略的文件

8.1Flume介紹FlumeSources

Taildir

Source用于觀察指定的文件，幾乎可以實時監(jiān)測到添加到每個文件的新行。如果文件正在寫入新行，則此采集器將重試采集它們以等待寫入完成，Source常用配置屬性如下所示。屬性名稱默認值相關(guān)說明channels–

type–組件類型名需必須是TAILDIRfilegroups–以空格分隔的文件組列表filegroups.<filegroupName>–文件組的絕對路徑idleTimeout120000關(guān)閉非活動文件的時間（毫秒）

8.1Flume介紹FlumeSources續(xù)表屬性名稱默認值相關(guān)說明writePosInterval3000寫入位置文件上每個文件的最后位置的間隔時間batchSize100一次讀取和發(fā)送到通道的最大行數(shù)backoffSleepIncrement1000當(dāng)最后一次嘗試未找到任何新數(shù)據(jù)時，每次重新嘗試輪詢新數(shù)據(jù)之間的最大時間延遲fileHeaderfalse是否添加存儲絕對路徑文件名的標頭fileHeaderKeyfile將絕對路徑文件名附加到eventheader時使用的header關(guān)鍵字

8.1Flume介紹FlumeSources續(xù)表屬性名稱默認值相關(guān)說明channels–

type

組件類型名需必須是httpport–采集源要綁定的端口bind要監(jiān)聽綁定的主機名或IP地址handlerorg.apache.flume.source.http.JSONHandlerhandler類的全路徑名

8.1Flume介紹FlumeSources

HTTPSource可以通過HTTPPOST和GET請求方式接收event數(shù)據(jù)，GET通常只能用于測試使用，POST請求發(fā)送的所有的events都被認為是一個批次，會在一個事務(wù)中插入channel，TaildirSource常用配置屬性如下所示。屬性名稱默認值相關(guān)說明channels–

type

組件類型名需必須是httpport–采集源要綁定的端口bind要監(jiān)聽綁定的主機名或IP地址handlerorg.apache.flume.source.http.JSONHandlerhandler類的全路徑名

8.1Flume介紹FlumeChannels

Channels通道是event在Agent上暫存的存儲庫，Source向Channel中添加event，Sink在讀取完數(shù)據(jù)后再刪除它。在配置Channels時，需要明確的就是將要傳輸?shù)膕ources數(shù)據(jù)源類型；根據(jù)這些信息結(jié)合開發(fā)中的實際需求，選擇Flume已提供的支持的FlumeChannels；再根據(jù)選擇的Channel類型，配置必要和非必要的Channel屬性，F(xiàn)lume提供并支持的FlumeChannels種類如下所示。MemoryChannelJDBCChannelKafkaChannelFileChannelSpillableMemoryChannelPseudoTransactionChannelCustomChannel

8.1Flume介紹FlumeChannels

MemoryChannel會將event存儲在具有可配置最大尺寸的內(nèi)存隊列中，適用于需要更高吞吐量的流量，但在Agent發(fā)生故障時會丟失部分階段數(shù)據(jù)，下表為MemoryChannel常用配置屬性。屬性名稱默認值相關(guān)說明type–組件類型名需必須是memorycapacity100存儲在channel中的最大event數(shù)transactionCapacity100channel從source接收或向sink傳遞的每個事務(wù)中最大event數(shù)keep-alive3添加或刪除event的超時時間(秒)byteCapacityBufferPercentage20定義byteCapacity與channel中所有event所占百分比byteCapacity等于JVM可用的最大內(nèi)存的80%允許此channel中所有event的的最大內(nèi)存字節(jié)數(shù)總和

8.1Flume介紹FlumeChannelsFileChannel是Flume的持久通道，它將所有event寫入磁盤，因此不會丟失進程或機器關(guān)機、崩潰時的數(shù)據(jù)。FileChannel通過在一次事務(wù)中提交多個event來提高吞吐量，做到了只要事務(wù)被提交，那么數(shù)據(jù)就不會有丟失，F(xiàn)ileChannel常用配置屬性如下所示。屬性名稱默認值相關(guān)說明type–組件類型名需必須是filecheckpointDir1~/.flume/file-channel/checkpoint檢測點文件所存儲的目錄useDualCheckpointsfalse備份檢測點如果設(shè)置為true，backupCheckpointDir必須設(shè)置backupCheckpointDir–備份檢查點目錄。此目錄不能與數(shù)據(jù)目錄或檢查點目錄相同dataDirs~/.flume/file-channel/data數(shù)據(jù)存儲所在的目錄設(shè)置

8.1Flume介紹FlumeChannels屬性名稱默認值相關(guān)說明transactionCapacity10000事務(wù)容量的最大值設(shè)置checkpointInterval30000檢測點之間的時間值設(shè)置（單位微秒）maxFileSize2146435071一個單一日志的最大值設(shè)置（以字節(jié)為單位）capacity1000000channel的最大容量transactionCapacity10000事務(wù)容量的最大值設(shè)置續(xù)表

8.1Flume介紹FlumeSinksFlumeSoures采集的數(shù)據(jù)通過Channels通道流向Sink中，此時Sink類似一個集結(jié)的遞進中心，它需要根據(jù)需求進行配置，從而最終選擇發(fā)送目的地。配置Sinks時，明確將要傳輸?shù)臄?shù)據(jù)目的地、結(jié)果類型；然后根據(jù)實際需求信息，選擇Flume已提供支持的FlumeSinks；再根據(jù)選擇的Sinks類型，配置必要和非必要的Sinks屬性。Flume提供并支持的FlumeSinks種類如下所示。HDFSSinkHiveSinkLoggerSinkAvroSinkThriftSinkIRCSinkFileRollSinkNullSinkHBaseSinkAsyncHBaseSinkMorphlineSolrSinkElasticSearchSinkKiteDatasetSinkKafkaSinkHTTPSinkCustomSink

8.1Flume介紹FlumeSinks

HDFSSink將event寫入Hadoop分布式文件系統(tǒng)（HDFS），它目前支持創(chuàng)建文本和序列文件，以及兩種類型的壓縮文件，下表為HDFSSink常用配置屬性。屬性名稱默認值相關(guān)說明channel–

type–組件類型名需必須是hdfshdfs.path–HDFS目錄路徑hdfs.filePrefixFlumeData為在hdfs目錄中由Flume創(chuàng)建的文件指定前綴hdfs.roundfalse是否應(yīng)將時間戳向下舍入

8.1Flume介紹FlumeSinks屬性名稱默認值相關(guān)說明hdfs.roundValue1舍入到此最高倍數(shù)，小于當(dāng)前時間hdfs.roundUnitsecond舍入值的單位-秒、分鐘或小時hdfs.rollInterval30滾動當(dāng)前文件之前等待的秒數(shù)hdfs.rollSize1024觸發(fā)滾動的文件大小，以字節(jié)為單位hdfs.rollCount10在滾動之前寫入文件的事件數(shù)續(xù)表

8.1Flume介紹FlumeSinks

LoggerSink用于記錄INFO級別event，它通常用于調(diào)試。LoggerSink接收器的不同之處是它不需要在“記錄原始數(shù)據(jù)”部分中說明額外的配置，LoggerSink常用配置屬性如下所示。屬性名稱默認值相關(guān)說明channel–

type–組件類型名需必須是loggermaxBytesToLog16要記錄的eventbody的最大字節(jié)數(shù)

8.1Flume介紹FlumeSinks

屬性名稱默認值相關(guān)說明channel–

type–組件類型名需必須是avrohostname–要監(jiān)聽的主機名或IP地址port–要監(jiān)聽的服務(wù)端口batch-size100要一起批量發(fā)送的event數(shù)connect-timeout20000允許第一次（握手）請求的時間量（ms）request-timeout20000在第一個之后允許請求的時間量（ms）AvroSink形成Flume分層收集支持的一半，發(fā)送到此接收器的Flumeevent轉(zhuǎn)換為Avroevent，并發(fā)送到對應(yīng)配置的主機名/端口，event將從配置的channel中批量獲取配置的批處理大小，AvroSink常用配置屬性如下所示。

8.1Flume介紹Flume日志采集系統(tǒng)結(jié)構(gòu)圖

在實際開發(fā)中，F(xiàn)lume需要采集數(shù)據(jù)的類型多種多樣，同時還會進行不同的中間操作，所以根據(jù)具體需求，可以將Flume日志采集系統(tǒng)分為單代理和多代理結(jié)構(gòu)。

8.1Flume介紹Flume日志采集系統(tǒng)結(jié)構(gòu)圖單代理Flume日志采集系統(tǒng)的結(jié)構(gòu)

8.1Flume介紹Flume日志采集系統(tǒng)結(jié)構(gòu)圖多代理的Flume日志采集系統(tǒng)的結(jié)構(gòu)

8.2Flume基本使用方法Flume系統(tǒng)要求

使用Flume進行開發(fā)，必須滿足一定的系統(tǒng)要求，這里以官方說明為準，具體要求如下。安裝Java1.8或更高版本Java運行環(huán)境；為Source（數(shù)據(jù)采集器）、Channel（緩沖通道）、Sink（接收器）的配置提供足夠的內(nèi)存空間；為Channel（緩沖通道）、Sink（接收器）的配置提供足夠的磁盤空間；保證Agent（代理）對要操作的目錄有讀寫權(quán)限.。

8.2Flume基本使用Flume安裝配置1下載Flume1.8.0安裝包并解壓。2配置flume-env.sh文件，添加JDK環(huán)境變量。3配置/etc/profile文件，添加Flume環(huán)境變量。

8.3采集方案配置說明

使用Flume系統(tǒng)，只需要創(chuàng)建一個配置文件，用來配置FlumeSource、FlumeChannel和FlumeSink三大組件的屬性即可。

8.3采集方案配置說明Flume

sources（1）AvrosourcesAvro是一個數(shù)據(jù)序列化的系統(tǒng),flume通過監(jiān)聽Avro端口，從外部Avroclient獲取events,配置文件格式如下：a1.sources=r1a1.channels=c1a1.sources.r1.type=avroa1.sources.r1.channels=c1a1.sources.r1.bind=a1.sources.r1.port=4141（2）Execsources此源啟動運行一個給定的Unix命令，預(yù)計這一過程中不斷產(chǎn)生標準輸出的數(shù)據(jù)。配置文件格式如下：exec-agent.sources=tailexec-agent.channels=memoryChannel-1exec-agent.sinks=loggerexec-agent.sources.tail.type=execmand=tail-f/var/log/secure

8.3采集方案配置說明Flumesources（3）Netcatsources一個netcat在某一端口上偵聽，每一行文字變成一個事件源。行為像“nc-k-l[主機][端口]”。換句話說，它打開一個指定端口，偵聽數(shù)據(jù)。意料的是，所提供的數(shù)據(jù)是換行符分隔的文本。每一行文字變成Flume事件，并通過連接通道發(fā)送。（4）SyslogTCPsources

用于監(jiān)控TCP端口信息，可以用來接收socket通信通過TCP發(fā)過來的信息。（5）SyslogUDPsources用于監(jiān)控UDP端口信息，可以用來接收socket通信通過TCP發(fā)過來的信息。

8.3采集方案配置說明FlumeChannel（1）MemoryChannel

用內(nèi)存空間來存儲sources收集到的數(shù)據(jù)，配置如下：

a1.channels=c1 a1.channels.c1.type=memory a1.channels.c1.capacity=10000 a1.channels.c1.transactionCapacity=10000 a1.channels.c1.byteCapacityBufferPercentage=20 a1.channels.c1.byteCapacity=800000（2）JDBCChannel用數(shù)據(jù)庫來存儲sources收集到的數(shù)據(jù)，目前支持DERBY數(shù)據(jù)庫（3）FILEChannel用文件當(dāng)做channel來存儲中間數(shù)據(jù)

8.3采集方案配置說明FlumeSinks（1）HDFSSinks用HDFS來存儲channel里面的信息，配置如下：

a1.channels=c1 a1.sinks=k1 a1.sinks.k1.type=hdfs a1.sinks.k1.channel=c1 a1.sinks.k1.hdfs.path=/flume/events/%y-%m-%d/%H%M/%S a1.sinks.k1.hdfs.filePrefix=events- a1.sinks.k1.hdfs.round=true a1.sinks.k1.hdfs.roundValue=10 a1.sinks.k1.hdfs.roundUnit=minute

8.3采集方案配置說明FlumeSinks（2）LoggerSinks將接收到的信息顯示在控制臺，配置如下：

a1.channels=c1 a1.sinks=k1 a1.sinks.k1.type=logger a1.sinks.k1.channel=c1（3）FileRollSinks將消息存儲在本地文件中，配置說明如下：a1.channels=c1a1.sinks=k1a1.sinks.k1.type=file_rolla1.sinks.k1.channel=c1a1.sinks.k1.sink.directory=/var/log/flume#文件存儲的路徑a1.sinks.k1.sink.rollInterval=30#多長時間往寫出一次。a1.sinks.k1.sink.serializer=TEXT#寫出格式

8.4Flume的可靠性保證負載均衡配置的采集方案是通過唯一一個Sink作為接收器接收后續(xù)需要的數(shù)據(jù)，但會出現(xiàn)當(dāng)前Sink故障或數(shù)據(jù)收集請求量較大的情況，這時單一Sink配置可能就無法保證Flume開發(fā)的可靠性。因此，F(xiàn)lume提供FlumeSinkProcessors解決上述問題。Sink處理器允許定義Sinkgroups，將多個sink分組到一個實體中，Sink處理器就可通過組內(nèi)多個sink為服務(wù)提供負載均衡功能。

8.4Flume的可靠性保證負載均衡負載均衡接收器處理器（Loadbalancingsinkprocessor）提供了在多個sink上進行負載均衡流量的功能，它維護一個活躍的sink索引列表，需在其上分配負載，還支持round_robin（輪詢）和random（隨機）選擇機制進行流量分配，默認選擇機制為round_robin。Loadbalancingsinkprocessor提供的配置屬性如下所示。屬性名稱默認值相關(guān)說明sinks–以空格分隔的參與sink組的sink列表processor.typedefault組件類型名需必須是load_balanceprocessor.backofffalse設(shè)置失敗的sink進入黑名單processor.selectorround_robin選擇機制processor.selector.maxTimeOut30000失敗sink放置在黑名單的超時時間

8.4Flume的可靠性保證故障轉(zhuǎn)移故障轉(zhuǎn)移接收器處理器（FailoverSinkProcessor）維護一個具有優(yōu)先級的sink列表，保證在處理event時，只需有一個可用的sink即可。故障轉(zhuǎn)移機制工作原理是將故障的sink降級到故障池中，在池中為它們分配一個冷卻期，在重試之前冷卻時間會增加，當(dāng)sink成功發(fā)送event后，它將恢復(fù)到活躍池中。FailoverSinkProcessor提供的配置屬性如下所示。屬性名稱默認值相關(guān)說明sinks–以空格分隔的參與sink組的sink列表processor.typedefault組件類型名需必須是failoverprocessor.priority.<sinkName>–設(shè)置sink的優(yōu)先級取值processor.maxpenalty30000失敗sink的最大退避時間

8.5Flume攔截器攔截器TimestampInterceptorHostInterceptorStaticInterceptorRemoveHeaderInterceptorUUIDInterceptorMorphlineInterceptorSearchandReplaceInterceptorRegexFilteringInterceptorRegexExtractorInterceptorFlumeInterceptors（攔截器）用于對Flume系統(tǒng)數(shù)據(jù)流中event的修改操作。使用Flume攔截器時，只需參考官方配置屬性在采集方案中選擇性的配置即可，當(dāng)涉及到配置多個攔截器時，攔截器名稱間需用空格分隔，且攔截器配置順序就是攔截順序。Flume1.8.0版本中，F(xiàn)lume提供并支持的攔截器有很多，具體如下所示。

8.5Flume攔截器攔截器TimestampInterceptor（時間戳攔截器）將流程執(zhí)行時間插入到event的header頭部，此攔截器插入帶有timestamp鍵的標頭，其值為對應(yīng)時間戳。若配置中已存在時間戳?xí)r，此攔截器可保留現(xiàn)有時間戳，TimestampInterceptor提供的常用配置屬性如下所示。屬性名稱默認值相關(guān)說明type–組件類型名需必須是timestampheadertimestamp用于放置生成的時間戳的標頭的名稱preserveExistingfalse如果時間戳已存在，是否應(yīng)保留,true或false

8.5Flume攔截器攔截器屬性名稱默認值相關(guān)說明type–組件類型名需必須是staticpreserveExistingtrue如果配置的header已存在，是否應(yīng)保留keykey應(yīng)創(chuàng)建的header的名稱valuevalue應(yīng)創(chuàng)建的header對應(yīng)的靜態(tài)值StaticInterceptor（靜態(tài)攔截器）允許用戶將具有靜態(tài)值的靜態(tài)頭附加到所有event。當(dāng)前不支持一次指定多個header頭，但是用戶可定義多個StaticInterceptor來為每一個攔截器都追加一個header，StaticInterceptor提供的常用配置屬性如下所示。

8.6案例——日志采集案例分析-數(shù)據(jù)抽取實時顯示目標：Flume監(jiān)控一端Console，另一端Console發(fā)送消息，使被監(jiān)控端實時顯示。分步實現(xiàn)：

創(chuàng)建FlumeAgent配置文件flume-telnet.conf安裝telnet工具3判斷44444端口是否被占用4先開啟flume先聽端口

5使用telnet工具向本機的44444端口發(fā)送內(nèi)容

8.6案例——日志采集案例分析-Hive日志上傳HDFS目標：實時監(jiān)控hive日志，并上傳到HDFS中分步實現(xiàn)：拷貝Hadoop相關(guān)jar到Flume的lib目錄下創(chuàng)建flume-hdfs.conf文件執(zhí)行監(jiān)控配置

8.6案例——日志采集案例分析-監(jiān)聽目錄的文件動態(tài)目標：實時監(jiān)控hive日志，并上傳到HDFS中分步實現(xiàn)：分步實現(xiàn)：創(chuàng)建配置文件flume-dir.conf執(zhí)行測試

8.6案例——日志采集多Agent數(shù)據(jù)傳遞-案例場景A、B兩臺日志服務(wù)機器實時生產(chǎn)日志主要類型為access.log、nginx.log、web.log現(xiàn)在要求：把A、B機器中的access.log、nginx.log、web.log采集匯總到C機器上然后統(tǒng)一收集到hdfs中。但是在hdfs中要求的目錄為：/source/logs/access/20160101/**/source/logs/nginx/20160101/**/source/logs/web/20160101/**

8.6案例——日志采集多Agent數(shù)據(jù)