public boolean aggregateJobDetails ( JobDetails jobDetails , AggregationConstants . AGGREGATION_TYPE aggType ) { Table aggTable = null ; try { switch ( aggType ) { case DAILY : aggTable = hbaseConnection . getTable ( TableName . valueOf ( AggregationConstants . AGG_DAILY_TABLE ) ) ; break ; case WEEKLY : aggTable = hbaseConnection . getTable ( TableName . valueOf ( AggregationConstants . AGG_WEEKLY_TABLE ) ) ; ; break ; default : LOG . error ( "Unknown aggregation type : " + aggType ) ; return false ; } // create row key JobKey jobKey = jobDetails . getJobKey ( ) ; AppAggregationKey appAggKey = new AppAggregationKey ( jobKey . getCluster ( ) , jobKey . getUserName ( ) , jobKey . getAppId ( ) , getTimestamp ( jobKey . getRunId ( ) , aggType ) ) ; LOG . info ( "Aggregating " + aggType + " stats for " + jobKey . toString ( ) ) ; Increment aggIncrement = incrementAppSummary ( appAggKey , jobDetails ) ; aggTable . increment ( aggIncrement ) ; boolean status = updateMoreAggInfo ( aggTable , appAggKey , jobDetails ) ; return status ; } /* * try to catch all exceptions so that processing is unaffected by * aggregation errors this can be turned off in the future when we determine * aggregation to be a mandatory part of Processing Step */ catch ( Exception e ) { LOG . error ( "Caught exception while attempting to aggregate for " + aggType + " table " , e ) ; return false ; } finally { if ( aggTable != null ) { try { aggTable . close ( ) ; } catch ( IOException e ) { LOG . error ( "Caught exception while attempting to close table " , e ) ; } } } }
creates a list of puts that aggregate the job details and stores in daily or weekly aggregation table