@ NotThreadsafe private void initFileStreams ( int chunkId ) { /** * {@link Set#add(Object)} returns false if the element already existed in the set. * This ensures we initialize the resources for each chunk only once. */ if ( chunksHandled . add ( chunkId ) ) { try { this . indexFileSizeInBytes [ chunkId ] = 0L ; this . valueFileSizeInBytes [ chunkId ] = 0L ; this . checkSumDigestIndex [ chunkId ] = CheckSum . getInstance ( checkSumType ) ; this . checkSumDigestValue [ chunkId ] = CheckSum . getInstance ( checkSumType ) ; this . position [ chunkId ] = 0 ; this . taskIndexFileName [ chunkId ] = new Path ( FileOutputFormat . getOutputPath ( conf ) , getStoreName ( ) + "." + Integer . toString ( chunkId ) + "_" + this . taskId + INDEX_FILE_EXTENSION + fileExtension ) ; this . taskValueFileName [ chunkId ] = new Path ( FileOutputFormat . getOutputPath ( conf ) , getStoreName ( ) + "." + Integer . toString ( chunkId ) + "_" + this . taskId + DATA_FILE_EXTENSION + fileExtension ) ; if ( this . fs == null ) this . fs = this . taskIndexFileName [ chunkId ] . getFileSystem ( conf ) ; if ( isValidCompressionEnabled ) { this . indexFileStream [ chunkId ] = new DataOutputStream ( new BufferedOutputStream ( new GZIPOutputStream ( fs . create ( this . taskIndexFileName [ chunkId ] ) , DEFAULT_BUFFER_SIZE ) ) ) ; this . valueFileStream [ chunkId ] = new DataOutputStream ( new BufferedOutputStream ( new GZIPOutputStream ( fs . create ( this . taskValueFileName [ chunkId ] ) , DEFAULT_BUFFER_SIZE ) ) ) ; } else { this . indexFileStream [ chunkId ] = fs . create ( this . taskIndexFileName [ chunkId ] ) ; this . valueFileStream [ chunkId ] = fs . create ( this . taskValueFileName [ chunkId ] ) ; } fs . setPermission ( this . taskIndexFileName [ chunkId ] , new FsPermission ( HadoopStoreBuilder . HADOOP_FILE_PERMISSION ) ) ; logger . info ( "Setting permission to 755 for " + this . taskIndexFileName [ chunkId ] ) ; fs . setPermission ( this . taskValueFileName [ chunkId ] , new FsPermission ( HadoopStoreBuilder . HADOOP_FILE_PERMISSION ) ) ; logger . info ( "Setting permission to 755 for " + this . taskValueFileName [ chunkId ] ) ; logger . info ( "Opening " + this . taskIndexFileName [ chunkId ] + " and " + this . taskValueFileName [ chunkId ] + " for writing." ) ; } catch ( IOException e ) { throw new RuntimeException ( "Failed to open Input/OutputStream" , e ) ; } } }
The MapReduce framework should operate sequentially so thread safety shouldn t be a problem .