基于Flume的美團(tuán)日志收集系統(tǒng)

無(wú)無(wú)好基于Flume的美團(tuán)日志收集系統(tǒng)(一)架構(gòu)和設(shè)計(jì)?2.2.架構(gòu)設(shè)計(jì)考慮需要考慮什么問(wèn)題3.3.Agent死機(jī)該如何解決4.4.Collector死機(jī)是否會(huì)有影響5.5.Flume-NG可靠性(reliability)方面做了哪些措施ume《基于Flume的美團(tuán)日志收集系統(tǒng)》將分兩部分給讀者呈現(xiàn)美團(tuán)日志收集系統(tǒng)的架構(gòu)設(shè)計(jì)第二部分改進(jìn)和優(yōu)化，將主要著眼于實(shí)際部署和使用過(guò)程中遇到的問(wèn)題，對(duì)Flume做的功系統(tǒng)簡(jiǎn)介系統(tǒng)所具有的基本特征。ook2常用的開源日志收集系統(tǒng)對(duì)比下面將對(duì)常見的開源日志收集系統(tǒng)Flume和Scribe的各方面進(jìn)行對(duì)比。對(duì)比中Flume將主ApacheFlumeNG們將常用的日志收集系統(tǒng)分為三層 (Agent層，Collector層和Store層)來(lái)進(jìn)行對(duì)比。對(duì)比項(xiàng)使用語(yǔ)言容錯(cuò)性載均衡可擴(kuò)展性Flume-NGJava好Scribec/c++Store之間有容錯(cuò)性；ent程度socket,text,tail等oletcp等代碼結(jié)構(gòu)系統(tǒng)框架好，模塊分明，易于開發(fā)代碼簡(jiǎn)單3美團(tuán)日志收集系統(tǒng)架構(gòu)Storm基于Flume設(shè)計(jì)和搭建而成。目前每下圖是美團(tuán)的日志收集系統(tǒng)的整體框架圖。gent臨時(shí)的日志存儲(chǔ)服務(wù)，或者將日志流導(dǎo)向其它服務(wù)器。r達(dá)到負(fù)載均衡的目標(biāo)，同時(shí)并處理單個(gè)Collector失效的問(wèn)題。后面介紹。kb.Channel統(tǒng)一使用美團(tuán)開發(fā)的DualChannel，具體原因后面詳述；對(duì)于過(guò)濾掉的日志使用NullChannel，具體原因后面詳述；4架構(gòu)設(shè)計(jì)考慮下面將從可用性，可靠性，可擴(kuò)展性和兼容性等方面，對(duì)上述的架構(gòu)做細(xì)致的解析。4.1可用性(availablity)對(duì)日志收集系統(tǒng)來(lái)說(shuō)，可用性(availablity)指固定周期內(nèi)系統(tǒng)無(wú)故障運(yùn)行總時(shí)間。要想提高統(tǒng)在可用性方面的考慮。4.1.1Agent死掉Agent種情況：機(jī)器死機(jī)或者Agent進(jìn)程死掉。對(duì)于機(jī)器死機(jī)的情況來(lái)說(shuō)，由于產(chǎn)生日志的進(jìn)程也同樣會(huì)死掉，所以不會(huì)再產(chǎn)生新的日志，不存在不提供服務(wù)的情況。提高系統(tǒng)的可用性。首先，所有的Agent在supervise的方式下啟動(dòng)，如果進(jìn)程死掉會(huì)被系A(chǔ)gent活監(jiān)控，發(fā)現(xiàn)Agent死掉立即報(bào)警。最后，對(duì)于非常重要的日志，建議應(yīng)用直接將日志寫磁盤，Agent使用spooldir的方式4.1.2Collector死掉由于中心服務(wù)器提供的是對(duì)等的且無(wú)差別的服務(wù)，且Agent訪問(wèn)Collector做了Collector服務(wù)不受影響。4.1.3Hdfs正常停機(jī)4.1.4Hdfs異常停機(jī)或不可訪問(wèn)Scribe，可以提供較好的容錯(cuò)性。4.1.5Collector變慢或者Agent/Collector網(wǎng)絡(luò)變慢如果Collector處理速度變慢(比如機(jī)器load過(guò)高)或者Agent/Collector之間的網(wǎng)絡(luò)變慢，4.1.6Hdfs變慢天，每周都有高峰使用期，所以這種情況非常普遍。4.2可靠性(reliability)這就是Flume提供數(shù)據(jù)流中點(diǎn)到點(diǎn)的可靠性保證的最基本的單跳消息傳遞語(yǔ)義。那么Flume是如何做到上述最基本的消息傳遞語(yǔ)義呢證了event在數(shù)據(jù)流的點(diǎn)對(duì)點(diǎn)傳輸中是可靠的。在多級(jí)數(shù)據(jù)流中，如下圖，上一級(jí)的Sink轉(zhuǎn)移。4.3可擴(kuò)展性(scalability)能夠以簡(jiǎn)單的增加機(jī)器來(lái)達(dá)到線性擴(kuò)容的目的。對(duì)于基于Flume的日志收集系統(tǒng)來(lái)說(shuō)，需要在設(shè)計(jì)的每一層，都可以做到線性擴(kuò)展地提供服務(wù)。下面將對(duì)每一層的可擴(kuò)展性做相應(yīng)的說(shuō)明。4.3.1Agent層4.3.2Collector層可以線性擴(kuò)展。其性能主要受限于Store層提供的能力。4.3.3Store層臨時(shí)的應(yīng)用，只對(duì)應(yīng)于某一類日志，性能不是瓶頸。4.4Channel的選擇MemoryChannel:所有的events被保存在內(nèi)存中。優(yōu)點(diǎn)是高吞吐。缺點(diǎn)是容量有限并FileChannel:所有的events被保存在文件中。優(yōu)點(diǎn)是容量較大且死掉時(shí)數(shù)據(jù)可恢復(fù)。缺點(diǎn)是速度較慢。Channel基于此，我們開發(fā)了DualChannel。DualChannel：基于MemoryChannel和FileChannel開發(fā)。當(dāng)堆積在Channel中的4.5和scribe兼容4.6權(quán)限控制入到存儲(chǔ)系統(tǒng)。所以目前的權(quán)限控制就是category過(guò)濾。修改麻煩，每增加一個(gè)日志就需要重啟或者重載Agent的配置。如果權(quán)限控制放在Collector端，優(yōu)勢(shì)是方便進(jìn)行配置的修改和加載。劣勢(shì)是部分沒有注冊(cè)r問(wèn)題屬于次要問(wèn)題，所以選擇采用Collector端控制。4.7提供實(shí)時(shí)流美團(tuán)的部分業(yè)務(wù)，如實(shí)時(shí)推薦，反爬蟲服務(wù)等服務(wù)，需要處理實(shí)時(shí)的數(shù)據(jù)流。因此我們希望一個(gè)非常重要的要求是實(shí)時(shí)數(shù)據(jù)流不應(yīng)該受到其它Sink的速度影響，保證實(shí)時(shí)數(shù)據(jù)流的速5系統(tǒng)監(jiān)控我們建立了多維度的監(jiān)控，防止未知的異常發(fā)生。5.1發(fā)送速度，擁堵情況，寫Hdfs速度于超預(yù)期的擁堵，我們會(huì)報(bào)警出來(lái)查找原因。們需要檢查是否有異常。這樣可以及時(shí)發(fā)現(xiàn)Flume和日志的異常.5.3日志大小異常監(jiān)控饋，幫助他們及早修復(fù)自身系統(tǒng)的異常。高可靠性，可擴(kuò)展等特性的分布式服務(wù)?；贔lume的美團(tuán)日志收集系統(tǒng)(二)改進(jìn)和優(yōu)化1.Flume的存在些什么問(wèn)題Flume加了哪些功能3.Flume系統(tǒng)如何調(diào)優(yōu)Flume》中，我們?cè)斒隽嘶贔lume的美團(tuán)和使用過(guò)程中遇到的問(wèn)題，對(duì)Flume的功能改進(jìn)和對(duì)系統(tǒng)做的優(yōu)化。Flume的問(wèn)題總結(jié)a.Channel“水土不服”：使用固定大小的MemoryChannel在日志高峰時(shí)常報(bào)隊(duì)列大小不夠c.系統(tǒng)的管理問(wèn)題：配置升級(jí)，模塊重啟等；2Flume的功能改進(jìn)和優(yōu)化點(diǎn)，修改了一些Bug，并且進(jìn)行一些調(diào)優(yōu)。下面將對(duì)一些主要的方面做一些說(shuō)明。2.1增加Zabbixmonitor服務(wù)dfselevent?Sink:已經(jīng)處理的event數(shù)tesink.batch-sizehdfs.batchSize縮格式總數(shù)據(jù)大平均events/s33333壓縮后大/***RenamebucketPathfilefrom.tmptopermanentlocation.privatevoidrenameBucket()throwsIOException,InterruptedException{if(bucketPath.equals(targetPath)){}finalPathsrcPath=newPath(bucketPath);finalPathdstPath=newPath(targetPath);callWithTimeout(newCallRunner<Object>(){@@OverridepublicObjectcall()throwsException{if(fileSystem.exists(srcPath)){//couldblockLOG.info("Renaming"+srcPath+"to"+dstPath);fileSystem.rename(srcPath,dstPath);//couldblock////indexthedstPathlzofileif(codeC!=null&&".lzo".equals(codeC.getDefaultExtension())){LzoIndexerLzoIndexerlzoIndexer=newLzoIndexer(newConfiguration());lzoIndexer.index(dstPath);}}returnnull;}}}2.3增加HdfsSink的開關(guān)2.4增加DualChannelChannel使用FileChannel，由l/****putToMemChannelindicateputeventtomemChannelorfileChannel*takeFromMemChannelindicatetakeeventfrommemChannelorfileChannelprivateAtomicBooleanputToMemChannel=newAtomicBoolean(true);privateprivateAtomicBooleantakeFromMemChannel=newAtomicBoolean(true);voiddoPut(Eventevent){if(switchon&&putToMemChannel.get()){lmemTransaction.put(event);ifif(memChannel.isFull()||fileChannel.getQueueSize()>100){putToMemChannel.set(false);}}}else{fileTransaction.put(event);}}EventEventdoTake(){Eventevent=null;ifif(takeFromMemChannel.get()){levent=memTransaction.take();ifif(event==null){takeFromMemChannel.set(false);}}else{event=fileTransaction.take();if(event==null){takeFromMemChanneltakeFromMemChannel.set(true);putToMemChannel.set(true);}}}returnevent;}2.5增加NullChannelink2.6增加KafkaSinkpublicclassKafkaSinkextendsAbstractSinkimplementsConfigurable{privateprivateStringzkConnect;privateIntegerzkTimeout;privateIntegerbatchSize;privateprivateIntegerqueueSize;privateStringserializerClass;privateprivateStringproducerType;privateStringtopicPrefix;privateProducer<String,String>producer;publicpublicvoidconfigure(Contextcontext){//讀取配置，并檢查配置}@Overridepublicsynchronizedvoidstart(){}@@Overridepublicsynchronizedvoidstop(){}@OverridepublicStatusprocess()throwsEventDeliveryException{Statusstatus=Status.READY;Channelchannel=getChannel();Transactiontx=channel.getTransaction();try{txtx.begin();Map<String,List<String>>topic2EventList=newHashMap<String,List<String>>();mit();}catch(Exceptione){tx.rollback();thrownewEventDeliveryException(e);}finally{tx.close();}returnstatus;}}2.7修復(fù)和scribe的兼容問(wèn)題3.Flume系統(tǒng)調(diào)優(yōu)經(jīng)驗(yàn)總結(jié)3.1基礎(chǔ)參數(shù)調(diào)優(yōu)經(jīng)驗(yàn)HdfsSink中默認(rèn)的serializer會(huì)每寫一行在行尾添加一個(gè)換行符，我們?nèi)罩颈旧韼в袚Q行符，這樣會(huì)導(dǎo)致每條日志后面多一個(gè)空行，修改配置不要自動(dòng)添加換行符；適當(dāng)調(diào)大HdfsSink的callTimeout，避免不必要的超時(shí)錯(cuò)誤；類的目錄，并且按天和按小時(shí)存放：me匹配快幾十倍。拼接定符串的方式，20萬(wàn)條日志的操作只需