001 /** 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018 019 package org.apache.hadoop.mapreduce.lib.output; 020 021 import java.io.IOException; 022 import java.text.NumberFormat; 023 024 import org.apache.hadoop.classification.InterfaceAudience; 025 import org.apache.hadoop.classification.InterfaceStability; 026 import org.apache.hadoop.conf.Configuration; 027 import org.apache.hadoop.fs.FileSystem; 028 import org.apache.hadoop.fs.Path; 029 import org.apache.hadoop.io.compress.CompressionCodec; 030 import org.apache.hadoop.mapred.FileAlreadyExistsException; 031 import org.apache.hadoop.mapred.InvalidJobConfException; 032 import org.apache.hadoop.mapreduce.Job; 033 import org.apache.hadoop.mapreduce.JobContext; 034 import org.apache.hadoop.mapreduce.OutputCommitter; 035 import org.apache.hadoop.mapreduce.OutputFormat; 036 import org.apache.hadoop.mapreduce.RecordWriter; 037 import org.apache.hadoop.mapreduce.TaskAttemptContext; 038 import org.apache.hadoop.mapreduce.TaskID; 039 import org.apache.hadoop.mapreduce.TaskInputOutputContext; 040 import org.apache.hadoop.mapreduce.security.TokenCache; 041 042 /** A base class for {@link OutputFormat}s that read from {@link FileSystem}s.*/ 043 @InterfaceAudience.Public 044 @InterfaceStability.Stable 045 public abstract class FileOutputFormat<K, V> extends OutputFormat<K, V> { 046 047 /** Construct output file names so that, when an output directory listing is 048 * sorted lexicographically, positions correspond to output partitions.*/ 049 private static final NumberFormat NUMBER_FORMAT = NumberFormat.getInstance(); 050 protected static final String BASE_OUTPUT_NAME = "mapreduce.output.basename"; 051 protected static final String PART = "part"; 052 static { 053 NUMBER_FORMAT.setMinimumIntegerDigits(5); 054 NUMBER_FORMAT.setGroupingUsed(false); 055 } 056 private FileOutputCommitter committer = null; 057 public static final String COMPRESS ="mapreduce.output.fileoutputformat.compress"; 058 public static final String COMPRESS_CODEC = 059 "mapreduce.output.fileoutputformat.compress.codec"; 060 public static final String COMPRESS_TYPE = "mapreduce.output.fileoutputformat.compress.type"; 061 public static final String OUTDIR = "mapreduce.output.fileoutputformat.outputdir"; 062 063 @Deprecated 064 public static enum Counter { 065 BYTES_WRITTEN 066 } 067 068 /** 069 * Set whether the output of the job is compressed. 070 * @param job the job to modify 071 * @param compress should the output of the job be compressed? 072 */ 073 public static void setCompressOutput(Job job, boolean compress) { 074 job.getConfiguration().setBoolean(FileOutputFormat.COMPRESS, compress); 075 } 076 077 /** 078 * Is the job output compressed? 079 * @param job the Job to look in 080 * @return <code>true</code> if the job output should be compressed, 081 * <code>false</code> otherwise 082 */ 083 public static boolean getCompressOutput(JobContext job) { 084 return job.getConfiguration().getBoolean( 085 FileOutputFormat.COMPRESS, false); 086 } 087 088 /** 089 * Set the {@link CompressionCodec} to be used to compress job outputs. 090 * @param job the job to modify 091 * @param codecClass the {@link CompressionCodec} to be used to 092 * compress the job outputs 093 */ 094 public static void 095 setOutputCompressorClass(Job job, 096 Class<? extends CompressionCodec> codecClass) { 097 setCompressOutput(job, true); 098 job.getConfiguration().setClass(FileOutputFormat.COMPRESS_CODEC, 099 codecClass, 100 CompressionCodec.class); 101 } 102 103 /** 104 * Get the {@link CompressionCodec} for compressing the job outputs. 105 * @param job the {@link Job} to look in 106 * @param defaultValue the {@link CompressionCodec} to return if not set 107 * @return the {@link CompressionCodec} to be used to compress the 108 * job outputs 109 * @throws IllegalArgumentException if the class was specified, but not found 110 */ 111 public static Class<? extends CompressionCodec> 112 getOutputCompressorClass(JobContext job, 113 Class<? extends CompressionCodec> defaultValue) { 114 Class<? extends CompressionCodec> codecClass = defaultValue; 115 Configuration conf = job.getConfiguration(); 116 String name = conf.get(FileOutputFormat.COMPRESS_CODEC); 117 if (name != null) { 118 try { 119 codecClass = 120 conf.getClassByName(name).asSubclass(CompressionCodec.class); 121 } catch (ClassNotFoundException e) { 122 throw new IllegalArgumentException("Compression codec " + name + 123 " was not found.", e); 124 } 125 } 126 return codecClass; 127 } 128 129 public abstract RecordWriter<K, V> 130 getRecordWriter(TaskAttemptContext job 131 ) throws IOException, InterruptedException; 132 133 public void checkOutputSpecs(JobContext job 134 ) throws FileAlreadyExistsException, IOException{ 135 // Ensure that the output directory is set and not already there 136 Path outDir = getOutputPath(job); 137 if (outDir == null) { 138 throw new InvalidJobConfException("Output directory not set."); 139 } 140 141 // get delegation token for outDir's file system 142 TokenCache.obtainTokensForNamenodes(job.getCredentials(), 143 new Path[] { outDir }, job.getConfiguration()); 144 145 if (outDir.getFileSystem(job.getConfiguration()).exists(outDir)) { 146 throw new FileAlreadyExistsException("Output directory " + outDir + 147 " already exists"); 148 } 149 } 150 151 /** 152 * Set the {@link Path} of the output directory for the map-reduce job. 153 * 154 * @param job The job to modify 155 * @param outputDir the {@link Path} of the output directory for 156 * the map-reduce job. 157 */ 158 public static void setOutputPath(Job job, Path outputDir) { 159 try { 160 outputDir = outputDir.getFileSystem(job.getConfiguration()).makeQualified( 161 outputDir); 162 } catch (IOException e) { 163 // Throw the IOException as a RuntimeException to be compatible with MR1 164 throw new RuntimeException(e); 165 } 166 job.getConfiguration().set(FileOutputFormat.OUTDIR, outputDir.toString()); 167 } 168 169 /** 170 * Get the {@link Path} to the output directory for the map-reduce job. 171 * 172 * @return the {@link Path} to the output directory for the map-reduce job. 173 * @see FileOutputFormat#getWorkOutputPath(TaskInputOutputContext) 174 */ 175 public static Path getOutputPath(JobContext job) { 176 String name = job.getConfiguration().get(FileOutputFormat.OUTDIR); 177 return name == null ? null: new Path(name); 178 } 179 180 /** 181 * Get the {@link Path} to the task's temporary output directory 182 * for the map-reduce job 183 * 184 * <h4 id="SideEffectFiles">Tasks' Side-Effect Files</h4> 185 * 186 * <p>Some applications need to create/write-to side-files, which differ from 187 * the actual job-outputs. 188 * 189 * <p>In such cases there could be issues with 2 instances of the same TIP 190 * (running simultaneously e.g. speculative tasks) trying to open/write-to the 191 * same file (path) on HDFS. Hence the application-writer will have to pick 192 * unique names per task-attempt (e.g. using the attemptid, say 193 * <tt>attempt_200709221812_0001_m_000000_0</tt>), not just per TIP.</p> 194 * 195 * <p>To get around this the Map-Reduce framework helps the application-writer 196 * out by maintaining a special 197 * <tt>${mapreduce.output.fileoutputformat.outputdir}/_temporary/_${taskid}</tt> 198 * sub-directory for each task-attempt on HDFS where the output of the 199 * task-attempt goes. On successful completion of the task-attempt the files 200 * in the <tt>${mapreduce.output.fileoutputformat.outputdir}/_temporary/_${taskid}</tt> (only) 201 * are <i>promoted</i> to <tt>${mapreduce.output.fileoutputformat.outputdir}</tt>. Of course, the 202 * framework discards the sub-directory of unsuccessful task-attempts. This 203 * is completely transparent to the application.</p> 204 * 205 * <p>The application-writer can take advantage of this by creating any 206 * side-files required in a work directory during execution 207 * of his task i.e. via 208 * {@link #getWorkOutputPath(TaskInputOutputContext)}, and 209 * the framework will move them out similarly - thus she doesn't have to pick 210 * unique paths per task-attempt.</p> 211 * 212 * <p>The entire discussion holds true for maps of jobs with 213 * reducer=NONE (i.e. 0 reduces) since output of the map, in that case, 214 * goes directly to HDFS.</p> 215 * 216 * @return the {@link Path} to the task's temporary output directory 217 * for the map-reduce job. 218 */ 219 public static Path getWorkOutputPath(TaskInputOutputContext<?,?,?,?> context 220 ) throws IOException, 221 InterruptedException { 222 FileOutputCommitter committer = (FileOutputCommitter) 223 context.getOutputCommitter(); 224 return committer.getWorkPath(); 225 } 226 227 /** 228 * Helper function to generate a {@link Path} for a file that is unique for 229 * the task within the job output directory. 230 * 231 * <p>The path can be used to create custom files from within the map and 232 * reduce tasks. The path name will be unique for each task. The path parent 233 * will be the job output directory.</p>ls 234 * 235 * <p>This method uses the {@link #getUniqueFile} method to make the file name 236 * unique for the task.</p> 237 * 238 * @param context the context for the task. 239 * @param name the name for the file. 240 * @param extension the extension for the file 241 * @return a unique path accross all tasks of the job. 242 */ 243 public 244 static Path getPathForWorkFile(TaskInputOutputContext<?,?,?,?> context, 245 String name, 246 String extension 247 ) throws IOException, InterruptedException { 248 return new Path(getWorkOutputPath(context), 249 getUniqueFile(context, name, extension)); 250 } 251 252 /** 253 * Generate a unique filename, based on the task id, name, and extension 254 * @param context the task that is calling this 255 * @param name the base filename 256 * @param extension the filename extension 257 * @return a string like $name-[mrsct]-$id$extension 258 */ 259 public synchronized static String getUniqueFile(TaskAttemptContext context, 260 String name, 261 String extension) { 262 TaskID taskId = context.getTaskAttemptID().getTaskID(); 263 int partition = taskId.getId(); 264 StringBuilder result = new StringBuilder(); 265 result.append(name); 266 result.append('-'); 267 result.append( 268 TaskID.getRepresentingCharacter(taskId.getTaskType())); 269 result.append('-'); 270 result.append(NUMBER_FORMAT.format(partition)); 271 result.append(extension); 272 return result.toString(); 273 } 274 275 /** 276 * Get the default path and filename for the output format. 277 * @param context the task context 278 * @param extension an extension to add to the filename 279 * @return a full path $output/_temporary/$taskid/part-[mr]-$id 280 * @throws IOException 281 */ 282 public Path getDefaultWorkFile(TaskAttemptContext context, 283 String extension) throws IOException{ 284 FileOutputCommitter committer = 285 (FileOutputCommitter) getOutputCommitter(context); 286 return new Path(committer.getWorkPath(), getUniqueFile(context, 287 getOutputName(context), extension)); 288 } 289 290 /** 291 * Get the base output name for the output file. 292 */ 293 protected static String getOutputName(JobContext job) { 294 return job.getConfiguration().get(BASE_OUTPUT_NAME, PART); 295 } 296 297 /** 298 * Set the base output name for output file to be created. 299 */ 300 protected static void setOutputName(JobContext job, String name) { 301 job.getConfiguration().set(BASE_OUTPUT_NAME, name); 302 } 303 304 public synchronized 305 OutputCommitter getOutputCommitter(TaskAttemptContext context 306 ) throws IOException { 307 if (committer == null) { 308 Path output = getOutputPath(context); 309 committer = new FileOutputCommitter(output, context); 310 } 311 return committer; 312 } 313 } 314