專18-1-實(shí)時(shí)保護(hù)Hadoop數(shù)據(jù)安全－蘇良飛

May2016?AboutEagle?Architecture?Ecosystem?Q&AApacheEagleisadistributedreal-timemonitoringandalertingengineforhadoopfromeBayOpensourcedasApacheIncubatorProjectonOct26th2015dataactivitymonitoringsolutiontoinstantlyidentifyaccesstosensitivedatarecognizeattacksmaliciousactivityandblockaccessinrealtime.ApacheIncubatoreagle.incubator.apacheApacheIHadoopEagleProductionReleaseDonatedtoApacheSoftwareFoundationASFfromeBayatOct6th,2015EEagleProjectInitiativeHadoopOctOct232015Oct262015GithubGithubOpenS/apache/incubator-eagle5HadoopData?Security?ActivityHadoopData?Security?ActivityHadoopPlatform?Heath?Availability?Performancetingtoollikezabbixgangliacannothandlethehugevolumeofmetrics/logsgeneratedbyhadoopsystemineBay.Hadoop@eBayInc3000+nodeses201020090+nodes20071-10nodes10,000nodes150,000+cores170PB20111000+nodes10,000+cores7Y7?AboutEagle?Architecture?Ecosystem?Q&AScalableScalestomonitorthousandsofpoliciesandbillionsofaccesseventsExtensibleEaglecanbeeasilyextendedtomonitorotherdatasourcesReal-timeGeneratesalertsinrealtimeandblocksuserswithmaliciousintentMachineLearningCreatedynamicuserprofilesbasedonuserbehaviorDataCollectorsEAMPROCESSINGPolicyEngineesholdsPolicyTADATAMANAGERDataCollectorsEAMPROCESSINGPolicyEngineesholdsPolicyTADATAMANAGERHDFS,Audit,SecurityCustommoduleApachest10?Real-timeDataCollection?DistributedPolicyEngine?StreamProcessingDSL?ScalableDataStorage&Query?MachineLearningIntegrationNOTE{NAME}-{NUMBER}likeHDFS-6914meansopensourceprojectticketidcontributedbyusDecouplingwithMessageBus?ApacheKafka:high-throughputdistributedmessaging?Partition:balancebetweenlogicandthroughputCross-PlatformIntegration?CommunityKafkaClient(18+)?Python/Go/C/C++/JAVA..?EnhancedLog4j-kafka?KAFKA-2041:ExtensiblePartitionKey?KAFKA-2077:AdvancedTopicSelector12FieldGroupingShuffleGroupingFieldGroupingShuffleGroupinglityFilebeatLogstashResourceconsumptionbalanceMessagethroughputbalance(LOGSTASH-179)Ligh-weightcollector(golang)withdaemonLogstashinstancesclusterDistributedMessageBusCentralizedStateManagementJOBLOGCentralizedStateManagementJOBLOGScalability:DistributedReal-timeIngestionpoutDistributedcrawlingforhadoopjobnodejmxandservicelogsetcReal-timeEventStreamvironmentagementReal-timeEventStreamvironmentagementTADATAMANAGER?Real-time?Usability?Scalability?Extensibility?Metadata-drivenAlertExecutor_{2}…DynamicalStreamSchemaDynamicalPolicyDeployment15agementDynamicalStreamSchematagementDynamicalStreamSchemat.Kafka-basedDistributedMessageBus(Extensible).Storm-basedReal-timeEventStream(EventStream(Kafka)(Extensible).StreameventsareprocessedandalertsareevaluatedduringstreamingTADATAMANAGER16Alerts?PowerfulSQL-LikeCEPCQLforPolicyAlerts?PowerfulSQL-LikeCEPCQLforPolicyDefinition?DynamicalPoilcyMetadataLifecycleManagement(Deployment/Update)?Easy-to-usePolicymanagementandAlertanalyticsUI(value>1000)]select*insertintooutputStream;agementagement17Full-functionStreamingCEPCQL:SiddhionStormbydefaultngaggValueinsertintooutputStream?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,StreamDefinitionandQueryPlancompilation18vironmentvironmentbilitydynamicpolicypartitionbyeventpolicy?NUserswith3partitions,Mpolicieswith2partitions,then3*2physicaltasks?Physicalpartition+policy-levelpartition…19DistributedStreamingPartitionProblemS={3,1,1,2,2,1,1}DistributedStreamingPartitionProblemS={3,1,1,2,2,1,1}S1={1,1,1,1,1}S2={2,2}S3={3}20Strategy?Greedy(Online/Offline)?PoTC?PKG?HashingStrategy?Greedy(Online/Offline)?PoTC?PKG?HashingicsCacheDistributedStreamingPartitionStrategygroupBygroupBy[GreedyStrategy]((_.key1,_.key2))KeyDistributionStatisticsfflineKeyDistributionStatisticsfflineStrategyExtensibility?ExtensiblepolicylifecyclemanagementExtensibility?Extensiblepolicylifecyclemanagement?SupportWSO2SiddhiCEPasfirstclass?ExtensiblepolicyengineimplementationpublicpublicinterfacePolicyEvaluatorServiceProvider{publicStringgetPolicyType();//literalstringtoidentifyonetypeofpolicypublicClassgetPolicyEvaluator();//getpolicyevaluatorimplementationpublicListgetBindingModules();//policytextwithjsonformattoobjectmapping}publicpublicinterfacePolicyEvaluator{publicvoidevaluate(ValuesArrayinput)throwsException;//evaluateinputeventpublicvoidonPolicyUpdate(AlertDefinitionAPIEntitynewAlertDef);//policyupdatepublicvoidonPolicyDelete();//invokedwhenpolicyisdeleted}DistributedReal-timePolicyEnginePolicy/MetadataTADATAMANAGER22n?PolicyDefinition:AlertDefinitionAPIEntity?Centralmetadatamanagementn?PolicyDefinition:AlertDefinitionAPIEntity?Centralmetadatamanagement?Dynamicmetadatadeployment?StreamSchema:AlertStreamSchemaEntity@@Table("alertdef")@ColumnFamily("f")@Prefix("alertdef")@Service(AlertConstants.ALERT_DEFINITION_SERVICE_ENDPOINT_NAME)@JsonIgnoreProperties(ignoreUnknown=true)@TimeSeries(false)@Tags({"site","dataSource","alertExecutorId","policyId","policyType"})@Indexes({@Index(name="Index_1_alertExecutorId",columns={"alertExecutorID"},unique=true),})publicclassAlertDefinitionAPIEntityextendsTaggedLogAPIEntity{@Column("a")privateStringdesc;@Column("b")privateStringpolicyDef;@Column("c")privateStringdedupeDef;DistributedDistributedReal-timePolicyEngineDynamicMetadataLoadingTADATAMANAGERStream_{1}StreamProcessingReal-timeEventStream Stream_{*}Stream_{1}StreamProcessingReal-timeEventStream Stream_{*}…usterEnvironmentfigOptimizer1.Development2.Optimization3.Compiletonativeapp?Physicalexecutionplatformindependent?Easilyassembledatatransformation,filtering,joinandalertingDAGinfluentway?DAGrewriteandoptimization?StreamUnionExpansion?StreamGroupbyExpansion?StreamNameExpansion?StreamAlertExpansion?StreamParallelismConfigExpansiontraittraitStreamProducer{filterflatMapmap{1,2,3,4}groupBystreamUnion//streamjoinishard,notimplementedforstormalertWithConsumer}StormExecutionEnvironmentenv=ExecutionEnvironmentFactory.getStorm(config);env.newSource(newKafkaSourcedSpoutProvider().getSpout(config)).renameOutputFields(1).flatMap(newAuditLogTransformer()).groupBy(0).flatMap(newUserProfileAggregatorExecutor());cutorenv.execute();25query=AlertDefinitionService[@dataSource="hiveQueryLog"]{@policyDef}query=AlertDefinitionService[@dataSource="hiveQueryLog"]{@policyDef}?EntityMetadataonlarge-scaleNoSQLstoragelikeHBase?Full-functionSQL-LikeRESTQuery?Optimizedrowkeydesignfortime-seriesmonitoringdataHBaseCoprocessor?SecondaryIndex@@Table("alertdef")@ColumnFamily("f")@Prefix("alertdef")@Service(AlertConstants.ALERT_DEFINITION_SERVICE_ENDPOINT_NAME)@JsonIgnoreProperties(ignoreUnknown=true)@TimeSeries(false)@Tags({"site","dataSource","alertExecutorId","policyId","policyType"})@Indexes({@Index(name="Index_1_alertExecutorId",columns={"alertExecutorID"},unique=true),})publicclassAlertDefinitionAPIEntityextendsTaggedLogAPIEntity{@Column("a")privateStringdesc;@Column("b")privateStringpolicyDef;@Column("c")privateStringdedupeDef;26mrowkeydesign…cRowkeyMetricNametimestamptagNametagValue…timestamptagNametagValue…Rowkey::=LogType|PartitionKeys|timestamp|tagName|tagValue|…Rowvalue::=LogContent27AnomalyMetricPredictiveDetectionCaseStudyAnomalyMetricPredictiveDetectionCaseStudyAnomalyMetricPredictiveDetectionOffline:Analyzingandcombining500+metricstogetherforcausalanomalydetections(IG->PCA->GMM->MCC)Online:PredictivelyalertforanomalymetricsPCAPrincipalComponentAnalysis)Normal(Green)andAbnormal(Red)DataandProbabilityDistributionandThresholdSelection?AboutApacheEagle?Architecture?Ecosystem?Q&AEagleFrameworkDistributedreal-timeframeworkforefficientlydevelopinghighlyscalablemonitoringapplicationsEagleAppsSecurity/Hadoop/DatabaseEagleInterfaceRESTService/ManagementUI/CustomizableAnalyticsVisualizationEagleIntegrationAmbari/Docker/Ranger/DataguiseOpenSourcetionInIntegrationàAmbariàDockeràRangeràDataguiseàWebPortalàRESTServicesàAnalyticsVisualizationAppsàSecurityàHadoopàCloudàDatabaseEagleFramework?HDFSAuditLog?HiveQueryLog?ApacheRanger?Adopteagleoutput?Dataguise?Cassandra?CapitialOne?Paypalserauserandsrcbsrcduserauserandsrcasrcorpc“->f=eventStream[cmd==‘rename’anduser==a.userandsrc==b.srcanddst==a.src]”GwedtrueugirootauthSIMPLEipcmdcreatesrctmpprivateCOPYINGINFOFSNamesystemauditallowedtrueugirootaut