public static Collection < WorkUnitState > mergeAllSplitWorkUnits ( FileSystem fs , Collection < WorkUnitState > workUnits ) throws IOException { ListMultimap < CopyableFile , WorkUnitState > splitWorkUnitsMap = ArrayListMultimap . create ( ) ; for ( WorkUnitState workUnit : workUnits ) { if ( isSplitWorkUnit ( workUnit ) ) { CopyableFile copyableFile = ( CopyableFile ) CopySource . deserializeCopyEntity ( workUnit ) ; splitWorkUnitsMap . put ( copyableFile , workUnit ) ; } } for ( CopyableFile file : splitWorkUnitsMap . keySet ( ) ) { log . info ( String . format ( "Merging split file %s." , file . getDestination ( ) ) ) ; WorkUnitState oldWorkUnit = splitWorkUnitsMap . get ( file ) . get ( 0 ) ; Path outputDir = FileAwareInputStreamDataWriter . getOutputDir ( oldWorkUnit ) ; CopyEntity . DatasetAndPartition datasetAndPartition = file . getDatasetAndPartition ( CopySource . deserializeCopyableDataset ( oldWorkUnit ) ) ; Path parentPath = FileAwareInputStreamDataWriter . getOutputFilePath ( file , outputDir , datasetAndPartition ) . getParent ( ) ; WorkUnitState newWorkUnit = mergeSplits ( fs , file , splitWorkUnitsMap . get ( file ) , parentPath ) ; for ( WorkUnitState wu : splitWorkUnitsMap . get ( file ) ) { // Set to committed so that task states will not fail wu . setWorkingState ( WorkUnitState . WorkingState . COMMITTED ) ; workUnits . remove ( wu ) ; } workUnits . add ( newWorkUnit ) ; } return workUnits ; }
Finds all split work units in the input collection and merges the file parts into the expected output files .