protected Collection < Path > getGranularInputPaths ( Path path ) throws IOException { boolean appendDelta = this . state . getPropAsBoolean ( MRCompactor . COMPACTION_RENAME_SOURCE_DIR_ENABLED , MRCompactor . DEFAULT_COMPACTION_RENAME_SOURCE_DIR_ENABLED ) ; Set < Path > uncompacted = Sets . newHashSet ( ) ; Set < Path > total = Sets . newHashSet ( ) ; for ( FileStatus fileStatus : FileListUtils . listFilesRecursively ( fs , path ) ) { if ( appendDelta ) { // use source dir suffix to identify the delta input paths if ( ! fileStatus . getPath ( ) . getParent ( ) . toString ( ) . endsWith ( MRCompactor . COMPACTION_RENAME_SOURCE_DIR_SUFFIX ) ) { uncompacted . add ( fileStatus . getPath ( ) . getParent ( ) ) ; } total . add ( fileStatus . getPath ( ) . getParent ( ) ) ; } else { uncompacted . add ( fileStatus . getPath ( ) . getParent ( ) ) ; } } if ( appendDelta ) { // When the output record count from mr counter doesn't match // the record count from input file names, we prefer file names because // it will be used to calculate the difference of count in next run. this . fileNameRecordCount = new InputRecordCountHelper ( this . state ) . calculateRecordCount ( total ) ; log . info ( "{} has total input record count (based on file name) {}" , path , this . fileNameRecordCount ) ; } return uncompacted ; }
Converts a top level input path to a group of sub - paths according to user defined granularity . This may be required because if upstream application generates many sub - paths but the map - reduce job only keeps track of the top level path after the job is done we won t be able to tell if those new arriving sub - paths is processed by previous map - reduce job or not . Hence a better way is to pre - define those sub - paths as input paths before we start to run MR . The implementation of this method should depend on the data generation granularity controlled by upstream . Here we just list the deepest level of containing folder as the smallest granularity .