java类org.apache.hadoop.io.compress.DefaultCodec的实例源码

protected SequenceFile.Writer getSequenceWriter(TaskAttemptContext context,
    Class<?> keyClass, Class<?> valueClass) 
    throws IOException {
  Configuration conf = context.getConfiguration();

  CompressionCodec codec = null;
  CompressionType compressionType = CompressionType.NONE;
  if (getCompressOutput(context)) {
    // find the kind of compression to do
    compressionType = getOutputCompressionType(context);
    // find the right codec
    Class<?> codecClass = getOutputCompressorClass(context, 
                                                   DefaultCodec.class);
    codec = (CompressionCodec) 
      ReflectionUtils.newInstance(codecClass, conf);
  }
  // get the path of the temporary output file 
  Path file = getDefaultWorkFile(context, "");
  FileSystem fs = file.getFileSystem(conf);
  return SequenceFile.createWriter(fs, conf, file,
           keyClass,
           valueClass,
           compressionType,
           codec,
           context);
}

/**
 * Get the codec string associated with this <code>configuration</code>
 *
 * @param configuration the {@link Configuration}
 * @return the codec {@link String}
 */
public static String getCodecString(Configuration configuration) {
  boolean compress = configuration.getBoolean(FileOutputFormat.COMPRESS, false);
  String codecType = configuration.get(FileOutputFormat.COMPRESS_TYPE, null);
  if (compress && (codecType == null || !codecType.equals(CompressionType.NONE.toString()))) {
    Class<?> codecClass = configuration.getClass(FileOutputFormat.COMPRESS_CODEC, DefaultCodec.class);
    if (codecClass == null) {
      return CODEC_NONE;
    } else {
      try {
        return ((CompressionCodec) codecClass.newInstance()).getDefaultExtension().replace(".", "");
      } catch (Exception exception) {
        throw new RuntimeException("Could not determine codec", exception);
      }
    }
  }
  return CODEC_NONE;
}

@Test
public void testGetCodecString() {
  Configuration configuration = dfsServer.getConf();
  assertEquals(MrUtil.CODEC_NONE, MrUtil.getCodecString(configuration));
  configuration.setBoolean(FileOutputFormat.COMPRESS, false);
  assertEquals(MrUtil.CODEC_NONE, MrUtil.getCodecString(configuration));
  configuration.setBoolean(FileOutputFormat.COMPRESS, true);
  assertEquals(new DefaultCodec().getDefaultExtension().substring(1, new DefaultCodec().getDefaultExtension().length()),
    MrUtil.getCodecString(configuration));
  configuration.set(FileOutputFormat.COMPRESS_CODEC, SnappyCodec.class.getName());
  assertEquals(new SnappyCodec().getDefaultExtension().substring(1, new SnappyCodec().getDefaultExtension().length()),
    MrUtil.getCodecString(configuration));
  configuration.set(FileOutputFormat.COMPRESS_TYPE, CompressionType.BLOCK.toString());
  assertEquals(new SnappyCodec().getDefaultExtension().substring(1, new SnappyCodec().getDefaultExtension().length()),
    MrUtil.getCodecString(configuration));
  configuration.set(FileOutputFormat.COMPRESS_TYPE, CompressionType.NONE.toString());
  assertEquals(MrUtil.CODEC_NONE, MrUtil.getCodecString(configuration));
  configuration.set(FileOutputFormat.COMPRESS_TYPE, CompressionType.BLOCK.toString());
  configuration.setBoolean(FileOutputFormat.COMPRESS, false);
  assertEquals(MrUtil.CODEC_NONE, MrUtil.getCodecString(configuration));
}

/**
 * For running a few tests of methods herein.
 * @param args
 * @throws IOException
 */
public static void main(String[] args) throws IOException {
  int count = 1024;
  int size = 10240;
  for (String arg: args) {
    if (arg.startsWith(COUNT)) {
      count = Integer.parseInt(arg.replace(COUNT, ""));
    } else if (arg.startsWith(SIZE)) {
      size = Integer.parseInt(arg.replace(SIZE, ""));
    } else {
      usage(1);
    }
  }
  IPCUtil util = new IPCUtil(HBaseConfiguration.create());
  ((Log4JLogger)IPCUtil.LOG).getLogger().setLevel(Level.ALL);
  timerTests(util, count, size,  new KeyValueCodec(), null);
  timerTests(util, count, size,  new KeyValueCodec(), new DefaultCodec());
  timerTests(util, count, size,  new KeyValueCodec(), new GzipCodec());
}

public DataOutputStream getDataOuputStream(final TaskAttemptContext job) throws IOException, InterruptedException {
    org.apache.hadoop.conf.Configuration hadoopConf = DEFAULT_COMPAT.getContextConfiguration(job);
    this.faunusConf = ModifiableHadoopConfiguration.of(hadoopConf);
    boolean isCompressed = getCompressOutput(job);
    CompressionCodec codec = null;
    String extension = "";
    if (isCompressed) {
        final Class<? extends CompressionCodec> codecClass = getOutputCompressorClass(job, DefaultCodec.class);
        codec = ReflectionUtils.newInstance(codecClass, hadoopConf);
        extension = codec.getDefaultExtension();
    }
    final Path file = super.getDefaultWorkFile(job, extension);
    final FileSystem fs = file.getFileSystem(hadoopConf);
    if (!isCompressed) {
        return new DataOutputStream(fs.create(file, false));
    } else {
        return new DataOutputStream(codec.createOutputStream(fs.create(file, false)));
    }
}

@Test
/**
 * Create an IFile.Writer using GzipCodec since this code does not
 * have a compressor when run via the tests (ie no native libraries).
 */
public void testIFileWriterWithCodec() throws Exception {
  Configuration conf = new Configuration();
  FileSystem localFs = FileSystem.getLocal(conf);
  FileSystem rfs = ((LocalFileSystem)localFs).getRaw();
  Path path = new Path(new Path("build/test.ifile"), "data");
  DefaultCodec codec = new GzipCodec();
  codec.setConf(conf);
  IFile.Writer<Text, Text> writer =
    new IFile.Writer<Text, Text>(conf, rfs.create(path), Text.class, Text.class,
                                 codec, null);
  writer.close();
}

@Test
/** Same as above but create a reader. */
public void testIFileReaderWithCodec() throws Exception {
  Configuration conf = new Configuration();
  FileSystem localFs = FileSystem.getLocal(conf);
  FileSystem rfs = ((LocalFileSystem)localFs).getRaw();
  Path path = new Path(new Path("build/test.ifile"), "data");
  DefaultCodec codec = new GzipCodec();
  codec.setConf(conf);
  FSDataOutputStream out = rfs.create(path);
  IFile.Writer<Text, Text> writer =
      new IFile.Writer<Text, Text>(conf, out, Text.class, Text.class,
                                   codec, null);
  writer.close();
  FSDataInputStream in = rfs.open(path);
  IFile.Reader<Text, Text> reader =
    new IFile.Reader<Text, Text>(conf, in, rfs.getFileStatus(path).getLen(),
        codec, null);
  reader.close();

  // test check sum 
  byte[] ab= new byte[100];
  int readed= reader.checksumIn.readWithChecksum(ab, 0, ab.length);
  assertEquals( readed,reader.checksumIn.getChecksum().length);

}

public static Job createJob(Path[] inputPaths, Path outputPath, Map<String, String> metadata, Configuration config)
        throws IOException
{
    final Job job = new Job(config);

    job.setInputFormatClass(SequenceFileInputFormat.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(BytesWritable.class);
    job.setOutputFormatClass(MetadataSequenceFileOutputFormat.class);

    SequenceFileInputFormat.setInputPaths(job, inputPaths);
    SequenceFileOutputFormat.setOutputPath(job, outputPath);
    SequenceFileOutputFormat.setCompressOutput(job, true);
    SequenceFileOutputFormat.setOutputCompressorClass(job, DefaultCodec.class);
    SequenceFileOutputFormat.setOutputCompressionType(job, CompressionType.BLOCK);

    if (metadata != null)
        MetadataConfiguration.setMetadata(metadata, job.getConfiguration());

    return job;
}

/** Unit tests for SequenceFile. */
@Test
public void testZlibSequenceFile() throws Exception {
  LOG.info("Testing SequenceFile with DefaultCodec");
  compressedSeqFileTest(new DefaultCodec());
  LOG.info("Successfully tested SequenceFile with DefaultCodec");
}

public void testValueIteratorWithCompression() throws Exception {
  Path tmpDir = new Path("build/test/test.reduce.task.compression");
  Configuration conf = new Configuration();
  DefaultCodec codec = new DefaultCodec();
  codec.setConf(conf);
  for (Pair[] testCase: testCases) {
    runValueIterator(tmpDir, testCase, conf, codec);
  }
}