private void aggreagteJobStats ( JobDetails jobDetails , byte [ ] rowKey , Context context , AggregationConstants . AGGREGATION_TYPE aggType ) throws IOException , InterruptedException { byte [ ] aggStatusCol = null ; switch ( aggType ) { case DAILY : aggStatusCol = AggregationConstants . JOB_DAILY_AGGREGATION_STATUS_COL_BYTES ; break ; case WEEKLY : aggStatusCol = AggregationConstants . JOB_WEEKLY_AGGREGATION_STATUS_COL_BYTES ; break ; default : LOG . error ( "Unknown aggregation type " + aggType ) ; return ; } boolean aggStatus = appSummaryService . aggregateJobDetails ( jobDetails , aggType ) ; context . progress ( ) ; LOG . debug ( "Status of aggreagting stats for " + aggType + "=" + aggStatus ) ; if ( aggStatus ) { // update raw table for this history file with aggregation status // Indicate that we processed the agg for this RAW successfully // so that we can skip it on the next scan (or not). Put aggStatusPut = rawService . getAggregatedStatusPut ( rowKey , aggStatusCol , aggStatus ) ; // TODO // In the unlikely event of multiple mappers running against one RAW // row, with one succeeding and one failing, // there could be a race where the // raw does not properly indicate the true status // (which is questionable in // any case with multiple simultaneous runs with different outcome). context . write ( RAW_TABLE , aggStatusPut ) ; } }
aggregate this job s stats only if re - aggregation is turned on OR aggreation is on AND job not already aggregated