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

無無好基于Flume的美團(tuán)日志收集系統(tǒng)(一)架構(gòu)和設(shè)計(jì)?2.2.架構(gòu)設(shè)計(jì)考慮需要考慮什么問題3.3.Agent死機(jī)該如何解決4.4.Collector死機(jī)是否會有影響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í)際部署和使用過程中遇到的問題，對Flume做的功系統(tǒng)簡介系統(tǒng)所具有的基本特征。ook2常用的開源日志收集系統(tǒng)對比下面將對常見的開源日志收集系統(tǒng)Flume和Scribe的各方面進(jìn)行對比。對比中Flume將主ApacheFlumeNG們將常用的日志收集系統(tǒng)分為三層 (Agent層，Collector層和Store層)來進(jìn)行對比。對比項(xiàng)使用語言容錯(cuò)性載均衡可擴(kuò)展性Flume-NGJava好Scribec/c++Store之間有容錯(cuò)性；ent程度socket,text,tail等oletcp等代碼結(jié)構(gòu)系統(tǒng)框架好，模塊分明，易于開發(fā)代碼簡單3美團(tuán)日志收集系統(tǒng)架構(gòu)Storm基于Flume設(shè)計(jì)和搭建而成。目前每下圖是美團(tuán)的日志收集系統(tǒng)的整體框架圖。gent臨時(shí)的日志存儲服務(wù)，或者將日志流導(dǎo)向其它服務(wù)器。r達(dá)到負(fù)載均衡的目標(biāo)，同時(shí)并處理單個(gè)Collector失效的問題。后面介紹。kb.Channel統(tǒng)一使用美團(tuán)開發(fā)的DualChannel，具體原因后面詳述；對于過濾掉的日志使用NullChannel，具體原因后面詳述；4架構(gòu)設(shè)計(jì)考慮下面將從可用性，可靠性，可擴(kuò)展性和兼容性等方面，對上述的架構(gòu)做細(xì)致的解析。4.1可用性(availablity)對日志收集系統(tǒng)來說，可用性(availablity)指固定周期內(nèi)系統(tǒng)無故障運(yùn)行總時(shí)間。要想提高統(tǒng)在可用性方面的考慮。4.1.1Agent死掉Agent種情況：機(jī)器死機(jī)或者Agent進(jìn)程死掉。對于機(jī)器死機(jī)的情況來說，由于產(chǎn)生日志的進(jìn)程也同樣會死掉，所以不會再產(chǎn)生新的日志，不存在不提供服務(wù)的情況。提高系統(tǒng)的可用性。首先，所有的Agent在supervise的方式下啟動，如果進(jìn)程死掉會被系A(chǔ)gent活監(jiān)控，發(fā)現(xiàn)Agent死掉立即報(bào)警。最后，對于非常重要的日志，建議應(yīng)用直接將日志寫磁盤，Agent使用spooldir的方式4.1.2Collector死掉由于中心服務(wù)器提供的是對等的且無差別的服務(wù)，且Agent訪問Collector做了Collector服務(wù)不受影響。4.1.3Hdfs正常停機(jī)4.1.4Hdfs異常停機(jī)或不可訪問Scribe，可以提供較好的容錯(cuò)性。4.1.5Collector變慢或者Agent/Collector網(wǎng)絡(luò)變慢如果Collector處理速度變慢(比如機(jī)器load過高)或者Agent/Collector之間的網(wǎng)絡(luò)變慢，4.1.6Hdfs變慢天，每周都有高峰使用期，所以這種情況非常普遍。4.2可靠性(reliability)這就是Flume提供數(shù)據(jù)流中點(diǎn)到點(diǎn)的可靠性保證的最基本的單跳消息傳遞語義。那么Flume是如何做到上述最基本的消息傳遞語義呢證了event在數(shù)據(jù)流的點(diǎn)對點(diǎn)傳輸中是可靠的。在多級數(shù)據(jù)流中，如下圖，上一級的Sink轉(zhuǎn)移。4.3可擴(kuò)展性(scalability)能夠以簡單的增加機(jī)器來達(dá)到線性擴(kuò)容的目的。對于基于Flume的日志收集系統(tǒng)來說，需要在設(shè)計(jì)的每一層，都可以做到線性擴(kuò)展地提供服務(wù)。下面將對每一層的可擴(kuò)展性做相應(yīng)的說明。4.3.1Agent層4.3.2Collector層可以線性擴(kuò)展。其性能主要受限于Store層提供的能力。4.3.3Store層臨時(shí)的應(yīng)用，只對應(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)限控制入到存儲系統(tǒng)。所以目前的權(quán)限控制就是category過濾。修改麻煩，每增加一個(gè)日志就需要重啟或者重載Agent的配置。如果權(quán)限控制放在Collector端，優(yōu)勢是方便進(jìn)行配置的修改和加載。劣勢是部分沒有注冊r問題屬于次要問題，所以選擇采用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ù)期的擁堵，我們會報(bào)警出來查找原因。們需要檢查是否有異常。這樣可以及時(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的存在些什么問題Flume加了哪些功能3.Flume系統(tǒng)如何調(diào)優(yōu)Flume》中，我們詳述了基于Flume的美團(tuán)和使用過程中遇到的問題，對Flume的功能改進(jìn)和對系統(tǒng)做的優(yōu)化。Flume的問題總結(jié)a.Channel“水土不服”：使用固定大小的MemoryChannel在日志高峰時(shí)常報(bào)隊(duì)列大小不夠c.系統(tǒng)的管理問題：配置升級，模塊重啟等；2Flume的功能改進(jìn)和優(yōu)化點(diǎn)，修改了一些Bug，并且進(jìn)行一些調(diào)優(yōu)。下面將對一些主要的方面做一些說明。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的兼容問題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會每寫一行在行尾添加一個(gè)換行符，我們?nèi)罩颈旧韼в袚Q行符，這樣會導(dǎo)致每條日志后面多一個(gè)空行，修改配置不要自動添加換行符；適當(dāng)調(diào)大HdfsSink的callTimeout，避免不必要的超時(shí)錯(cuò)誤；類的目錄，并且按天和按小時(shí)存放：me匹配快幾十倍。拼接定符串的方式，20萬條日志的操作只需