Apache Eagle-eBay構(gòu)建開源分布式實(shí)時(shí)預(yù)警引擎實(shí)踐-陳浩

ApacheEagle:eBay構(gòu)建開源分布式實(shí)時(shí)預(yù)警平臺(tái)實(shí)踐htp://~hao2016-4-232016-4-23ApacheEagle聯(lián)合發(fā)起人,PMC&CommittereBay分析平臺(tái)基礎(chǔ)架構(gòu)部門高級(jí)工程師全球Hadoop峰會(huì)(SJC全球Hadoop峰會(huì)(SJC/SHA/BJ/SZ)特邀講師htp://~hao諸多開源項(xiàng)目貢獻(xiàn)者h(yuǎn)tps://haoch?Eagle簡介?Eagle技術(shù)架構(gòu)?Eagle應(yīng)用場(chǎng)景?關(guān)于開源?Q&AIncludingApacheEagleandMoreLISTINGSCREATEDLISTINGSCREATEDVIAMOBILEEVERYWEEKGMVVIAMOBILEMOBILEMOBILEAPPDOWNLOADSGLOBALLYACTIVEBUYERS2525MACTIVESELLERSACTIVELISTINGSNEWLISTINGSEVERYWEEK可信賴的商業(yè)基礎(chǔ)架構(gòu)平臺(tái)APPLIATIONS/DATABASEMetricEventCriticalEventVisioVision?Availability?SecurityCapability?Monitoring?AlertingApacheEagle于2015年10月26日被接受入Apache孵化器http://Real-timeAnomalyDetectioninBigDataSecurity作為第一個(gè)組件目前主要專注于實(shí)p及時(shí)的觸發(fā)修復(fù)措施了解更多請(qǐng)參考http:/或http://apache/incubator-eagleHadoopData?HadoopData?Security?ActivityHadoopPlatform?Heath?Availability?PerformanceEaglewasinitializedbyendof2013forhadoopecosystemmonitoringasanyexistingtoollikezabbix,gangliacannothandlethehugevolumeofmetrics/logsgeneratedbyhadoopsystemineBay.HadoopHadoop@eBayInc3000+nodes10,000+cores50+PB20111000+nodes10,000+cores10+PB2014/201510,000nodes150,000+cores200950+nodes2012 理和存儲(chǔ)?大量數(shù)據(jù)吞吐(實(shí)時(shí)數(shù)據(jù)收集以及流處理IO復(fù)雜度)?復(fù)雜的異常檢測(cè)規(guī)則(流處理計(jì)算復(fù)雜度，Window等內(nèi)存空間復(fù)雜度)?多類型semi-structure數(shù)據(jù)存儲(chǔ)和查詢(Event,Log,Metric) 2動(dòng)態(tài)預(yù)警策略(Policy)和動(dòng)態(tài)關(guān)聯(lián)模型?復(fù)雜可描述的策略模型(SQLonStreaming)?動(dòng)態(tài)的關(guān)聯(lián)規(guī)則(動(dòng)態(tài)流式Sort,GroupBy,Join,Window)?機(jī)器學(xué)習(xí)預(yù)警模型 3Hadoop生態(tài)系統(tǒng)集成?動(dòng)態(tài)實(shí)時(shí)數(shù)據(jù)源管理(動(dòng)態(tài)管理KafkaTopic)?Hadoop原生預(yù)警策略(Policy)集合 4多租戶平臺(tái)支持?資源調(diào)度與隔離(獨(dú)立可用級(jí)別保證)?高可用與擴(kuò)展(彈性的資源管理)?近百條安全策略?8物理主機(jī)?30工作進(jìn)程?64kafka分區(qū)Eagle性能?平均延遲(Latency):~50ms?單集群最大吞吐量(Throughput):300k/stsAlertsReal-timeEventStreamPolicytsAlertsReal-timeEventStreamPolicy.實(shí)時(shí)流處理(Streaming):ApacheStorm(ExecutionEngine)+Kafka(MessageBus).描述性預(yù)警策略(Policy):SQL(CEP)onStreaming+動(dòng)態(tài)部署管理.線性擴(kuò)展(Scalability):數(shù)據(jù)量擴(kuò)展＋計(jì)算擴(kuò)展.元數(shù)據(jù)驅(qū)動(dòng)(Metadata):Schema管理和動(dòng)態(tài)協(xié)同AlertExecutorAlertExecutor_{1}AlertExecutor_{2}…AlertExecutor_{N}mStream_{1}Stream_{*}DynamicalStreamSchemaDynamicalPolicyDeploymentMETADATAMETADATAMANAGERAlertExecutor_{1}AlertExecutor_{2}…AlertExecutor_{N}mStreamAlertExecutor_{1}AlertExecutor_{2}…AlertExecutor_{N}mStream_{1}Stream_{*}tsAlertsDynamicalStreamSchemaDynamicalPolicyDeploymentMETADATAMANAGERPolicy.實(shí)時(shí)流處理(Streaming):ApacheStorm(ExecutionEngine)+Kafka(MessageBus).描述性預(yù)警策略(Policy):SQL(CEP)onStreaming+動(dòng)態(tài)部署管理.線性擴(kuò)展(Scalability):數(shù)據(jù)量擴(kuò)展＋計(jì)算擴(kuò)展.元數(shù)據(jù)驅(qū)動(dòng)(Metadata):Schema管理和動(dòng)態(tài)協(xié)同fromfromMetricStreamnameReplLagand(value>1000)]select*insertintooutputStream;RealReal-timeEventStreameplLagandvalueselectinsertintooutputStream;基于SQL的分布式流式處理:SiddhiCEP+Stormbydefault??Filter?Join?Aggregation:Avg,Sum,Min,Max,etc?Groupby?Having?Streamhandlersforwindow:TimeWindow,BatchWindow,LengthWindow?ConditionsandExpressions:and,or,not,==,!=,>=,>,<=,<,andarithmeticoperations?PatternProcessing?Sequenceprocessing?EventTables:intergratehistoricaldatainrealtimeprocessing?SQL-LikeQuery:Query,StreamDefinitionandQueryPlancompilation/wso2/siddhi示例1:Alertifhadoopnamenodecapacityusageexceed90percentagesfromhadoopJmxMetricEventStream[metric=="node.fsnamesystemstate.capacityused"andvalue>0.9]selectmetric,host,value,timestamp,component,siteinsertintoalertStream;示例2:AlertifhadoopnamenodeHAswitchesfromeverya=hadoopJmxMetricEventStream[metric=="node.fsnamesystem.hastate"]->b=hadoopJmxMetricEventStream[metric==a.metricandb.host==a.hostanda.value!=value)]within10minselecta.host,a.valueasoldHaState,b.valueasnewHaState,b.timestampastimestamp,b.metricasmetric,ponentascomponent,b.siteassiteinsertintoalertStream;DistributedStreamingClusterEnvironmentDistributedStreamingClusterEnvironment線性伸縮原理AlertExecutorAlertExecutor_{1}AlertExecutor_{2}…AlertExecutor_{N}mStream_{1}Stream_{*}分流不均衡問題/wiki/Partition_problemStreamPartitionSkew(15:1)gorithmeightsofExecutorsByPartitionUsergDistributedRealgDistributedReal-timePolicyEngineyiCEP策策略引擎的擴(kuò)展性體現(xiàn)在：?WSO2SiddhiCEPasfirstcitizen?ExtensiblePolicyEngineImplementation?ExtensiblePolicyLifecycleManagement?Metadata-basedModuleManagementMETADATAMETADATAMANAGERpublicinterfacePolicyEvaluatorServiceProvider{publicStringgetPolicyType();//literalstringtoidentifyonetypeofpolicypublicClassgetPolicyEvaluator();//getpolicyevaluatorimplementationpublicListgetBindingModules();//policytextwithjsonformattoobjectmapping}高度抽象的EagleDSL調(diào)用AlertLibrary1.Development2.Optimization3.Compiletonativeapp??輕量級(jí)ORM框架支持HBase和RDBMS?功能強(qiáng)大的統(tǒng)一類描述性RESTWebService接口?針對(duì)監(jiān)控?cái)?shù)據(jù)特征進(jìn)行了特別優(yōu)化的存儲(chǔ)結(jié)構(gòu)和Rokwey?提供原生Coprocessor支持，極大降低大數(shù)據(jù)量聚合時(shí)的延遲?應(yīng)用層二級(jí)索引的支持____e"policyType"})alertExecutorIDuniquetrue),ublicclassAlertDefinitionAPIEntityextendsIEntityprivateStringdesc;rivateStringpolicyDefrivateStringdedupeDefQueryQuery=AlertDefinitionService[@dataSource="hiveQueryLog"]{@policyDef}統(tǒng)一針對(duì)監(jiān)控?cái)?shù)據(jù)的HBaseRowkey設(shè)計(jì)Rowkey::=Prefix|PartitionKeys|timestamp|tagName|tagValue|…?MetricRowkey::=MetricName|PartitionKeys|timestamp|tagName|tagValue|…?EntityRowkey::=DefaultPrefix|PartitionKeys|timestamp|tagName|tagValue|…?LogRowkey::=LogType|PartitionKeys|timestamp|tagName|tagValue|…Rowvalue::=LogContent?元數(shù)據(jù)驅(qū)動(dòng)動(dòng)態(tài)Topology管理?動(dòng)態(tài)數(shù)據(jù)源(KafkaTopic)加載?動(dòng)態(tài)GroupBy/Join路由?動(dòng)態(tài)Policy部署與執(zhí)行?可定制預(yù)警發(fā)布/通知?資源調(diào)度和隔離?資源隔離單元：CEPRuntime,Bolt,Topology,Cluster?通過實(shí)時(shí)統(tǒng)計(jì)不同資源單位性能指標(biāo)(IO,CPU,Memory)以動(dòng)態(tài)優(yōu)化Policy分布?高可用和容錯(cuò)?流式計(jì)算狀態(tài)管理：MessageWAL+Siddhi(CheckpointSnapshot)+Storm狀態(tài)管理?HA:通過多Topology實(shí)例實(shí)現(xiàn)HA，同時(shí)支持無宕機(jī)維護(hù)或升級(jí)?可伸縮性?IO伸縮：針對(duì)單一數(shù)據(jù)源基于PartitionKey分流?計(jì)算伸縮：針對(duì)Policy計(jì)算復(fù)雜度或者內(nèi)存消耗分布?跨Topology伸縮：不同Topology僅執(zhí)行部分Policy?跨集群伸縮：不同Cluster僅執(zhí)行部分(Scalability)或全部Policy(DR)?單一Runtime關(guān)聯(lián)?Sort,Groupby,Join,Window?Policy動(dòng)態(tài)部署(HotDeploy)?生命周期管理(Lifecycle)?單一數(shù)據(jù)流多重關(guān)聯(lián)?單一數(shù)據(jù)流進(jìn)行多種不同GroupBy?單一數(shù)據(jù)流進(jìn)行多種不同Sort?單一數(shù)據(jù)流進(jìn)行多種不同Join?多數(shù)據(jù)流多重關(guān)聯(lián)?多數(shù)據(jù)流(Stream)JOIN?實(shí)時(shí)(Real-time)與歷史(Historical)數(shù)據(jù)流JOIN?Alert多重關(guān)聯(lián)去燥1大數(shù)據(jù)安全：實(shí)時(shí)異常數(shù)據(jù)行為安全監(jiān)控instantlyidentifyaccesstousactivityandblockaccessinrealtime12Hadoop性能監(jiān)控：JoB性能監(jiān)控與異常檢測(cè)hAnomalyDetection24 3eBay全球統(tǒng)一監(jiān)控系統(tǒng)平臺(tái)預(yù)警引擎Sharedmulti-tenantalertengineofglobalunifiedmonitoringplatform 4其他通用分布式實(shí)時(shí)AnomalyDetection/Alerting場(chǎng)景PrivilegesColumnsDataSetsPrivilegesColumnsDataSets大數(shù)據(jù)安全：異常數(shù)據(jù)行為安全監(jiān)控?cái)?shù)據(jù)丟失保護(hù)oususertryingtoster異常登錄或授權(quán)malicioususertriestoguesspasswordEaglecreatesuserprofilesusingchinelearningalgorithmtodetectanomalies未授權(quán)訪問tryingtoaccessclassifieddatawithoutprivilege異常數(shù)據(jù)操作ryingtodeletelargeamountofdataOperationtypeisoneeterofEagleuserprofilesEaglesupportsmultiplenativeoperationtypes.CCommonUserQueryPatternsQueryZonesCommandsZonesDetectDetectanomaliesinaccessingHDFSandHive§HDFS監(jiān)控策略§Hive監(jiān)控策略eHive區(qū)域(Zone)§數(shù)據(jù)安全分類以及敏感表識(shí)基于機(jī)器學(xué)習(xí)的用戶畫像(UserProfile)離線訓(xùn)練離線訓(xùn)練:Determinebandwidthfromtrainingdatasetthekerneldensityfunctionparameters(KDE)在線探測(cè):Ifatestdatapointliesoutsidethetrainedbandwidth,itisanomaly(Policy)PCs(PrincipleComponents)inEVD(EigenvalueValueDecomposition)KernelDensityFunctionHadoop性能監(jiān)控：Job/Node異常監(jiān)測(cè)場(chǎng)場(chǎng)景Detectnodeanomalybyanalyzingtaskfailureratioacrossallnodes假設(shè)Taskfailureratioforeverynodeshouldbeapproximatelyequal算法Nodebynodecompare(symmetryviolation)andpernodetrend基于Task統(tǒng)計(jì)模型的節(jié)點(diǎn)異常預(yù)警與分析預(yù)警:AnomalyDetection預(yù)警:AnomalyDetectionAlerting分析:Taskfailuredrill-downnModeling&StatisticsAvgMModeling&StatisticsAvgMinMaxDistributionsMaxz-scoreCorrelationCounters&FeaturesontionapInputRecordsduceInputRecordsInputRecordsmapSpilledRecordseduceShuffleRecordsapLocalFileBytesReadduceLocalFileBytesReadmapHDFSBytesReadreduceHDFSBytesRead場(chǎng)場(chǎng)景Detectdataskewbystatisticsanddistributionsforattemptexecutiondurationsandcounters假設(shè)DurationandcountersshouldbeinnormaldistributionThresholdThreshold&DetectionCorrelation>0.9&Max(Z-Score)>90%Counterseration1Eagle框架1Distributedreal-timeframeworkforefficientlydevelopinghighlyscalablemonitoringapplications2àAmbarià2àAmbariàDockeràRangeràDataguiseAp