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
019package org.apache.hadoop.mapreduce.lib.output;
020
021import java.io.IOException;
022import java.text.NumberFormat;
023
024import org.apache.hadoop.classification.InterfaceAudience;
025import org.apache.hadoop.classification.InterfaceStability;
026import org.apache.hadoop.conf.Configuration;
027import org.apache.hadoop.fs.FileSystem;
028import org.apache.hadoop.fs.Path;
029import org.apache.hadoop.io.compress.CompressionCodec;
030import org.apache.hadoop.mapred.FileAlreadyExistsException;
031import org.apache.hadoop.mapred.InvalidJobConfException;
032import org.apache.hadoop.mapreduce.Job;
033import org.apache.hadoop.mapreduce.JobContext;
034import org.apache.hadoop.mapreduce.OutputCommitter;
035import org.apache.hadoop.mapreduce.OutputFormat;
036import org.apache.hadoop.mapreduce.RecordWriter;
037import org.apache.hadoop.mapreduce.TaskAttemptContext;
038import org.apache.hadoop.mapreduce.TaskID;
039import org.apache.hadoop.mapreduce.TaskInputOutputContext;
040import 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
045public 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;
057public static final String COMPRESS ="mapreduce.output.fileoutputformat.compress";
058public static final String COMPRESS_CODEC = 
059"mapreduce.output.fileoutputformat.compress.codec";
060public static final String COMPRESS_TYPE = "mapreduce.output.fileoutputformat.compress.type";
061public 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   * <b id="SideEffectFiles">Tasks' Side-Effect Files</b>
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