private void archive ( Path parentPath , List < Path > srcPaths , Path outputPath , boolean append ) throws IOException { parentPath = parentPath . makeQualified ( parentPath . getFileSystem ( conf ) ) ; checkPaths ( conf , srcPaths ) ; Path destinationDir = outputPath . getParent ( ) ; FileOutputFormat . setOutputPath ( conf , outputPath ) ; FileSystem outFs = outputPath . getFileSystem ( conf ) ; if ( append ) { if ( ! outFs . exists ( outputPath ) ) { throw new IOException ( "Invalid Output. HAR File " + outputPath + "doesn't exist" ) ; } if ( outFs . isFile ( outputPath ) ) { throw new IOException ( "Invalid Output. HAR File " + outputPath + "must be represented as directory" ) ; } } else { if ( outFs . exists ( outputPath ) ) { throw new IOException ( "Invalid Output: " + outputPath + ". File already exists" ) ; } if ( outFs . isFile ( destinationDir ) ) { throw new IOException ( "Invalid Output. " + outputPath + " is not a directory" ) ; } } long totalSize = writeFilesToProcess ( parentPath , srcPaths ) ; FileSystem fs = parentPath . getFileSystem ( conf ) ; conf . set ( SRC_LIST_LABEL , srcFiles . toString ( ) ) ; conf . set ( SRC_PARENT_LABEL , parentPath . makeQualified ( fs ) . toString ( ) ) ; conf . setLong ( TOTAL_SIZE_LABEL , totalSize ) ; long partSize = conf . getLong ( HAR_PARTSIZE_LABEL , HAR_PARTSIZE_DEFAULT ) ; int numMaps = ( int ) ( totalSize / partSize ) ; //run atleast one map. conf . setNumMapTasks ( numMaps == 0 ? 1 : numMaps ) ; conf . setNumReduceTasks ( 1 ) ; conf . setOutputFormat ( NullOutputFormat . class ) ; conf . setMapOutputKeyClass ( IntWritable . class ) ; conf . setMapOutputValueClass ( Text . class ) ; conf . set ( "hadoop.job.history.user.location" , "none" ) ; //make sure no speculative execution is done conf . setSpeculativeExecution ( false ) ; if ( append ) { // set starting offset for mapper int partId = findFirstAvailablePartId ( outputPath ) ; conf . setInt ( PART_ID_OFFSET , partId ) ; Path index = new Path ( outputPath , HarFileSystem . INDEX_NAME ) ; Path indexDirectory = new Path ( outputPath , HarFileSystem . INDEX_NAME + ".copy" ) ; outFs . mkdirs ( indexDirectory ) ; Path indexCopy = new Path ( indexDirectory , "data" ) ; outFs . rename ( index , indexCopy ) ; MultipleInputs . addInputPath ( conf , jobDirectory , HArchiveInputFormat . class , HArchivesMapper . class ) ; MultipleInputs . addInputPath ( conf , indexDirectory , TextInputFormat . class , HArchivesConvertingMapper . class ) ; conf . setReducerClass ( HArchivesMergingReducer . class ) ; } else { conf . setMapperClass ( HArchivesMapper . class ) ; conf . setInputFormat ( HArchiveInputFormat . class ) ; FileInputFormat . addInputPath ( conf , jobDirectory ) ; conf . setReducerClass ( HArchivesReducer . class ) ; } JobClient . runJob ( conf ) ; cleanJobDirectory ( ) ; }
archive the given source paths into the dest