Flume數(shù)據(jù)采集課程目標本課程的目的是幫助學員了解flume數(shù)據(jù)采集的作用及工作原理，掌握flume的安裝配置，深入了解flume中source、channel、sink各個組件支持的常用類型，能夠使用flume進行數(shù)據(jù)的采集，能夠進行負載均衡、故障恢復、通道的多路復用，使用Flume攔截器進行數(shù)據(jù)的過濾。課程內容Flume概述Flume使用通道接收器源與通道選擇器攔截器Flume簡介Flume是Cloudera公司提供的分布式、可靠、高可用的海量日志采集、聚合和傳輸?shù)娜罩臼占到y(tǒng)，支持在日志系統(tǒng)中定制各類數(shù)據(jù)發(fā)送方，用于收集數(shù)據(jù)，flume也提供對數(shù)據(jù)進行簡單處理，并寫到各種數(shù)據(jù)接收方。Flume已被捐贈給了apache軟件基金會，為hadoop相關組件之一。應用場景一個電子商務網(wǎng)站，想分析消費者的行為或者購買意圖，快速的將他想要商品推送到界面上，要實現(xiàn)這一功能，我們需要對他曾經(jīng)訪問的頁面以及點擊的產(chǎn)品數(shù)據(jù)等日志數(shù)據(jù)信息進行收集并移交給平臺進行分析。Flume可以幫助我們進行數(shù)據(jù)的收集?，F(xiàn)在流行的內容推送，比如廣告定點投放以及新聞私人定制等等，很多場景都可以使用日志采集工具進行數(shù)據(jù)的收集。與Flume類似的工具：facebook的Scribe、Apache的chukwa、淘寶TimeTunnel。Flume的特征Flume可以高效率的將多個網(wǎng)站服務器中收集的日志信息存入HDFS/HBase中除了日志信息，F(xiàn)lume也可以用來接入收集規(guī)模宏大的社交網(wǎng)絡節(jié)點事件數(shù)據(jù)，比如facebook,twitter,電商網(wǎng)站如亞馬遜等支持各種接入資源數(shù)據(jù)的類型以及接出數(shù)據(jù)類型支持多路徑流量，多管道接入流量，多管道接出流量，上下文路由等可以被水平擴展Flume架構對于每一個Agent來說,它就是一個獨立的守護進程(JVM),它從客戶端那里接收數(shù)據(jù),或者從其他的Agent那里接收數(shù)據(jù),然后迅速的將獲取的數(shù)據(jù)傳給下一個目的節(jié)點sink,或者agent.

Agent主要由:source、channel、sink三個組件組成。Flume事件事件作為Flume內部數(shù)據(jù)傳輸?shù)淖罨締卧?，它是由轉載數(shù)據(jù)的字節(jié)數(shù)組(該數(shù)據(jù)組是從數(shù)據(jù)源接入點傳入，并傳輸給傳輸器，也就是HDFS/HBase)和可選頭部構成.Flume插件Interceptors攔截器用來在source和channel之間更改或者檢查Flume的events數(shù)據(jù)管道選擇器channelsSelectors在多管道時，被用來選擇使用那一條管道來傳遞數(shù)據(jù)(events).

管道選擇器又分為如下兩種:復制管道選擇器（默認的）:

每一個管道傳遞的都是相同的events多路通道選擇器:根據(jù)某些頭部信息將事件寫入到不同的通道中課程內容Flume概述Flume使用通道接收器源與通道選擇器攔截器Flume安裝配置下載Flumeapache-flume-1.8.0-bin.tar.gz解壓tar-xzvfapache-flume-1.8.0-bin.tar.gz測試安裝是否正確運行flume-ngversion，

輸出版本信息表示安裝正確Flume支持的類型SourcesChannelsSinksAvroSourceThriftSourceExecSourceJMSSourceSpoolingDirectorySourceTwitter1%firehoseSourceKafkaSourceNetCatSourceSequenceGeneratorSourceSyslogSourcesSyslogTCPSourceMultiportSyslogTCPSourceSyslogUDPSourceHTTPSourceStressSourceLegacySourcesThriftLegacySourceCustomSourceScribeSourceMemoryChannelJDBCChannelKafkaChannelFileChannelSpillableMemoryChannelPseudoTransactionChannelHDFSSinkHiveSinkLoggerSinkAvroSinkThriftSinkIRCSinkFileRollSinkNullSinkHBaseSinkAsyncHBaseSinkMorphlineSolrSinkElasticSearchSinkKiteDatasetSinkKafkaSinkflume示例創(chuàng)建netcat.conf文件從整體上描述代理agent中sources、sinks、channels所涉及到的組件agent.sources=s1agent.sinks=k1agent.channels=c1詳細描述agent中每一個source、sink與channel的具體實現(xiàn)agent.sources.s1.type=netcatagent.sources.s1.bind=localhostagent.sources.s1.port=8888agent.channels.c1.type=memoryagent.sinks.k1.type=loggerflume示例通過channel將source與sink連接起來agent.sources.s1.channels=c1agent.sinks.k1.channel=c1啟動flumeagent服務flume-ngagent-nagent-c../conf-f../conf/netcat.conf-Dflume.root.logger=INFO,console在另一個終端寫數(shù)據(jù)telnet8888課程內容Flume概述Flume使用通道接收器源與通道選擇器攔截器通道—內存通道內存通道指的是事件存儲在內存中的通道。通常情況下,內存的速度要比磁盤快幾個數(shù)量級,因此事件的接收速度也會更快,這降低了對硬件的需求量。使用這種通道的弊端在于代理失敗(如硬件問題、斷電、JVM崩潰、Flume重啟等)會導致數(shù)據(jù)丟失。系統(tǒng)度量通常屬于這一類,因為少量的數(shù)據(jù)丟失并不會造成什么影響。如果事件表示的是網(wǎng)站的購買情況,那么使用內存通道就是一種非常差勁的選擇了。使用內存通道,需要將通道的type參數(shù)設定為memory：agent.channels.c1.type=memory通道—內存通道配置參數(shù)通道—文件通道文件通道指的是將事件存儲到代理本地文件系統(tǒng)中的通道。雖然要比內存通道慢一些,不過它卻提供了持久化的存儲路徑,可以集的現(xiàn)應對大多數(shù)情況,它用在數(shù)據(jù)流中不允許出現(xiàn)缺口的場合。這種持久化能力是由WriteAheadLog（WAL)以及一個或多個文件存儲目錄聯(lián)合提供的。WAL用于以一種原子且安全的方式追蹤來自于通道的所有輸入與輸出。通過這種方式,如果代理重啟,那么WAL可以重放,從而確保在清理本地文件系統(tǒng)的數(shù)據(jù)存儲前進入到通道中的所有事件都會被寫出。agent.channels.c1.type=file通道—文件通道配置參數(shù)課程內容Flume概述Flume使用通道接收器源與通道選擇器攔截器HDFS接收器HDFS接收器的作用是持續(xù)打開HDFS中的文件，然后以流的方式將數(shù)據(jù)寫入，并且在某個時間點關閉該文件，再打開新的文件。agent.sinks.k1.type=hdfsagent.sinks.k1.hdfs.path=/outputagent.sinks.k1.hdfs.filePrefix=testagent.sinks.k1.hdfs.fileSuffix=.logagent.sinks.k1.hdfs.rollInterval=60agent.sinks.k1.hdfs.rollSize=0agent.sinks.k1.hdfs.rollCount=0事件序列化器事件序列化器是指flume事件轉換為另一種格式進行輸出的機制默認的序列化器是text序列化器，只會輸出flume事件體而丟棄頭信息可以通過text_with_headers序列化器保存flume事件頭和體，輸出格式包含了頭信息，后跟一個空格，然后是體信息負載。serializer=text_with_headers文件類型--SequenceFilesHDFS接受器將數(shù)據(jù)寫到hdfs中，默認的文件類型是HadoopSequenceFiles的形式。每個記錄的鍵是一個LongWritable,包含了當前的時間戳。值的格式默認是個與byte[]Flume體相對應的org.apache.Hadoop.io.Bytewritable文件類型--DataStream/CompressedStream如果數(shù)據(jù)沒有自然的鍵，而且不想輸出為SequenceFiles，那么可以使用DataStream數(shù)據(jù)流，只輸出未壓縮的值。agent.sinks.k1.hdfs.fileType=DataStreamagent.sinks.k1.hdfs.writeFormat=TextCompressedStream類似于DataStream，只是其數(shù)據(jù)在寫入時會被壓縮agent.sinks.k1.hdfs.fileType=CompressedStreamagent.sinks.k1.hdfs.codeC=gzip接收器組為了在數(shù)據(jù)處理管道中消除單點失敗,Flume提供了通過負載均衡或是故障恢復機制將事件發(fā)送到不同接收器的能力，需要使用接收器組。接收器組用于創(chuàng)建邏輯接收器分組,該分組的行為是由接收處理器來控制的,它決定了事件的路由方式。為了讓Flume知道接收器組,Fiume需要使用sinkgroups屬性agent.sinkgroups=sg1agent.sinkgroups.sg1.sinks=k1,k2負載均衡負載均衡負載均衡需要指定的屬性processor.type設為load_balance,就會使用循環(huán)選擇processor.selector屬性設定負載均衡算法：round_robin、random.故障恢復事件被發(fā)送到一個優(yōu)先的sink，如果這個優(yōu)先的sink不可用，則切換到備用的sink。agent.sinkgroups.sg1.sinks=k1,k2cessor.type=failovercessor.priority.k1=10cessor.priority.k1=20課程內容Flume概述Flume使用通道接收器源與通道選擇器攔截器Exec源Exec源基于Unix的command，收集運行命令輸出的結果作為它的數(shù)據(jù)源，將標準輸出中的每一行轉化成一個事件。agent.sources.s1.type=exec通過channel將source連接起來agent.sources.s1.channels=c1設置command屬性mand=tail-F/home/hadoop/app.logSpoolingDirectory源SpoolingDirectorySource：監(jiān)聽一個指定的目錄，只要應用程序向這個指定的目錄中添加新的文件，source組件就可以獲取到該信息，并解析該文件的內容，然后寫入到channle。寫入完成后，標記該文件已完成或者刪除該文件。agent.sources.s1.type=spooldiragent.sources.s1.spoolDir=/home/hadoop/datainputavro源avro是序列化的一種，實現(xiàn)了RPC（RemoteProcedureCall），avro用于支持節(jié)點與節(jié)點之間大批量數(shù)據(jù)交換的應用。avro支持二進制序列化方式，可以便捷、快速地處理大量數(shù)據(jù)的傳輸，數(shù)據(jù)存儲格式是JSON。avro數(shù)據(jù)傳輸?shù)牧鞒虉D：HttpSource有些應用程序環(huán)境，可能不能部署FlumeSDK及其依賴項，或客戶端代碼傾向于通過HTTP而不是Flume的PRC發(fā)送數(shù)據(jù)的情況，此時HTTPSOURCE可以用來將數(shù)據(jù)接收到Flume中。flume自帶的HttpSource可以通過HttpPost接收事件a1.sources.r1.type=httpa1.sources.r1.port=50000a1.sources.r1.host=配置參數(shù)描述typehttp(org.apache.fluem.source.httpSource）bind綁定的IP地址或主機名port綁定的端口號通道選擇器—復制一個源可以寫到一個或多個通道中，處理多個通道的方式有兩種：事件寫到多個通道中根據(jù)某個flume頭值寫到一個通道中Flume的這種內部機制叫通道選擇器，可以通過selector.type屬性指定任意通道的選擇器。在默認情況下，如果沒有為源指定選擇器，那么其默認值為replicating．replicating選擇器會將相同的事件寫到源的通道列表的所有通道中。agent.sources.s1.channels=c1c2agent.sinks.k1.channel=c1agent.sinks.k2.channel=c2agent.sources.s1.selector.type=replicating通道選擇器—多路復用如果想向不同的通道發(fā)送不同的事件，那就應該使用多路復用通道選擇器，方式是將selector.type設為multiplexing。事件要聲明一個頭部，然后我們檢查頭部對應的值，如果指定的值與設定的通道相匹配，那么就將該事件發(fā)送到被匹配到的通道中去。需要告訴通道選擇器使用哪個頭屬性，方式是設置selector.header屬性.agent.sources.s1.selector.type=multiplexinga1.sources.r1.selector.header=statea1.sources.r1.selector.mapping.CZ=c1a1.sources.r1.selector.mapping.US=c2a1.sources.r1.selector.default=c1課程內容Flume概述Flume使用通道接收器源與通道選擇器攔截器Flume攔截器攔截器（interceptor）是簡單的插件式組件，設置在source和channel之間。source接收到的事件，在寫入channel之前，攔截器可以利用eventsheader中加入的一些有用信息，或者對events的內容進行過濾，完成初步的數(shù)據(jù)清洗，在實際業(yè)務場景中非常有用。Flume自帶的攔截器TimestampInterceptor：當前時間攔截器

HostInterceptor：主機名攔截器

StaticInterceptor:可以在event的header中添加自定義的key和value

RegexFilteringInterceptor:通過正則表達式來清洗或包含匹配的events

RegexExtractorInterceptor:通過正則表達式將事件體的內容抽取出來放到flume頭中以便通過通道選擇器執(zhí)行路由。TimestampInterceptor時間戳攔截器，將當前時間戳（毫秒）加入到eventsheader中，key名字為：timestamp，值為當前時間戳。timestamp攔截器的屬性erceptors=i1erceptors.i1.type=timestampHostInterceptor主機名攔截器將運行Flumeagent的主機名或者IP地址加入到eventsheader中，key名字為：hosthost攔截器的屬性erceptors=i1erceptors.i1.type=hosterceptors.i1.hostHeader=hostnameerceptors.i2.useIP

=

falsea1.sinks.k1.hdfs.filePrefix

=

%{hostname}StaticInterceptor靜態(tài)攔截器用于在eventsheader中加入單個靜態(tài)的key和value，如果需要多個鍵/值，添加多個static攔截器。static攔截器的屬性：erceptors=i1erceptors.i1.type=staticerceptors.i1.key=usererceptors.i1.value=ljsRegexFilteringInterceptorRegexFilteringInterceptor攔截器用于過濾事件，篩選出與配置的正則表達式相匹配的事件?？梢杂糜诎?/p>