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

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