SequenceFileOutputFormat.java
上传用户:quxuerui
上传日期:2018-01-08
资源大小:41811k
文件大小:4k
源码类别:

网格计算

开发平台:

Java

  1. /**
  2.  * Licensed to the Apache Software Foundation (ASF) under one
  3.  * or more contributor license agreements.  See the NOTICE file
  4.  * distributed with this work for additional information
  5.  * regarding copyright ownership.  The ASF licenses this file
  6.  * to you under the Apache License, Version 2.0 (the
  7.  * "License"); you may not use this file except in compliance
  8.  * with the License.  You may obtain a copy of the License at
  9.  *
  10.  *     http://www.apache.org/licenses/LICENSE-2.0
  11.  *
  12.  * Unless required by applicable law or agreed to in writing, software
  13.  * distributed under the License is distributed on an "AS IS" BASIS,
  14.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15.  * See the License for the specific language governing permissions and
  16.  * limitations under the License.
  17.  */
  18. package org.apache.hadoop.mapred;
  19. import java.io.IOException;
  20. import java.util.Arrays;
  21. import org.apache.hadoop.fs.FileSystem;
  22. import org.apache.hadoop.fs.Path;
  23. import org.apache.hadoop.fs.FileUtil;
  24. import org.apache.hadoop.io.SequenceFile;
  25. import org.apache.hadoop.io.SequenceFile.CompressionType;
  26. import org.apache.hadoop.io.compress.CompressionCodec;
  27. import org.apache.hadoop.io.compress.DefaultCodec;
  28. import org.apache.hadoop.conf.Configuration;
  29. import org.apache.hadoop.util.*;
  30. /** An {@link OutputFormat} that writes {@link SequenceFile}s. 
  31.  * @deprecated Use 
  32.  *   {@link org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat} 
  33.  *   instead.
  34.  */
  35. @Deprecated
  36. public class SequenceFileOutputFormat <K,V> extends FileOutputFormat<K, V> {
  37.   public RecordWriter<K, V> getRecordWriter(
  38.                                           FileSystem ignored, JobConf job,
  39.                                           String name, Progressable progress)
  40.     throws IOException {
  41.     // get the path of the temporary output file 
  42.     Path file = FileOutputFormat.getTaskOutputPath(job, name);
  43.     
  44.     FileSystem fs = file.getFileSystem(job);
  45.     CompressionCodec codec = null;
  46.     CompressionType compressionType = CompressionType.NONE;
  47.     if (getCompressOutput(job)) {
  48.       // find the kind of compression to do
  49.       compressionType = getOutputCompressionType(job);
  50.       // find the right codec
  51.       Class<? extends CompressionCodec> codecClass = getOutputCompressorClass(job,
  52.   DefaultCodec.class);
  53.       codec = ReflectionUtils.newInstance(codecClass, job);
  54.     }
  55.     final SequenceFile.Writer out = 
  56.       SequenceFile.createWriter(fs, job, file,
  57.                                 job.getOutputKeyClass(),
  58.                                 job.getOutputValueClass(),
  59.                                 compressionType,
  60.                                 codec,
  61.                                 progress);
  62.     return new RecordWriter<K, V>() {
  63.         public void write(K key, V value)
  64.           throws IOException {
  65.           out.append(key, value);
  66.         }
  67.         public void close(Reporter reporter) throws IOException { out.close();}
  68.       };
  69.   }
  70.   /** Open the output generated by this format. */
  71.   public static SequenceFile.Reader[] getReaders(Configuration conf, Path dir)
  72.     throws IOException {
  73.     FileSystem fs = dir.getFileSystem(conf);
  74.     Path[] names = FileUtil.stat2Paths(fs.listStatus(dir));
  75.     
  76.     // sort names, so that hash partitioning works
  77.     Arrays.sort(names);
  78.     
  79.     SequenceFile.Reader[] parts = new SequenceFile.Reader[names.length];
  80.     for (int i = 0; i < names.length; i++) {
  81.       parts[i] = new SequenceFile.Reader(fs, names[i], conf);
  82.     }
  83.     return parts;
  84.   }
  85.   /**
  86.    * Get the {@link CompressionType} for the output {@link SequenceFile}.
  87.    * @param conf the {@link JobConf}
  88.    * @return the {@link CompressionType} for the output {@link SequenceFile}, 
  89.    *         defaulting to {@link CompressionType#RECORD}
  90.    */
  91.   public static CompressionType getOutputCompressionType(JobConf conf) {
  92.     String val = conf.get("mapred.output.compression.type", 
  93.                           CompressionType.RECORD.toString());
  94.     return CompressionType.valueOf(val);
  95.   }
  96.   
  97.   /**
  98.    * Set the {@link CompressionType} for the output {@link SequenceFile}.
  99.    * @param conf the {@link JobConf} to modify
  100.    * @param style the {@link CompressionType} for the output
  101.    *              {@link SequenceFile} 
  102.    */
  103.   public static void setOutputCompressionType(JobConf conf, 
  104.                                           CompressionType style) {
  105.     setCompressOutput(conf, true);
  106.     conf.set("mapred.output.compression.type", style.toString());
  107.   }
  108. }