public Result verify ( FileSystemDataset dataset ) { Map < String , Double > thresholdMap = RecompactionConditionBasedOnRatio . getDatasetRegexAndRecompactThreshold ( state . getProp ( MRCompactor . COMPACTION_LATEDATA_THRESHOLD_FOR_RECOMPACT_PER_DATASET , StringUtils . EMPTY ) ) ; CompactionPathParser . CompactionParserResult result = new CompactionPathParser ( state ) . parse ( dataset ) ; double threshold = RecompactionConditionBasedOnRatio . getRatioThresholdByDatasetName ( result . getDatasetName ( ) , thresholdMap ) ; log . debug ( "Threshold is {} for dataset {}" , threshold , result . getDatasetName ( ) ) ; InputRecordCountHelper helper = new InputRecordCountHelper ( state ) ; try { double newRecords = helper . calculateRecordCount ( Lists . newArrayList ( new Path ( dataset . datasetURN ( ) ) ) ) ; double oldRecords = helper . readRecordCount ( new Path ( result . getDstAbsoluteDir ( ) ) ) ; if ( oldRecords == 0 ) { return new Result ( true , "" ) ; } if ( ( newRecords - oldRecords ) / oldRecords > threshold ) { log . debug ( "Dataset {} records exceeded the threshold {}" , dataset . datasetURN ( ) , threshold ) ; return new Result ( true , "" ) ; } return new Result ( false , String . format ( "%s is failed for dataset %s. Prev=%f, Cur=%f, not reaching to threshold %f" , this . getName ( ) , result . getDatasetName ( ) , oldRecords , newRecords , threshold ) ) ; } catch ( IOException e ) { return new Result ( false , ExceptionUtils . getFullStackTrace ( e ) ) ; } }
There are two record count we are comparing here 1 ) The new record count in the input folder 2 ) The record count we compacted previously from last run Calculate two numbers difference and compare with a predefined threshold .