Author: cdouglas
Date: Mon Sep 14 00:45:01 2009
New Revision: 814455
URL: http://svn.apache.org/viewvc?rev=814455&view=rev
Log:
HADOOP-5879. Read compression level and strategy from Configuration for
gzip compression. Contributed by He Yongqiang
Modified:
hadoop/common/trunk/CHANGES.txt
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/CodecPool.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/Compressor.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/GzipCodec.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/bzip2/BZip2DummyCompressor.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/zlib/BuiltInZlibDeflater.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/zlib/ZlibCompressor.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/zlib/ZlibFactory.java
hadoop/common/trunk/src/test/core/org/apache/hadoop/io/compress/TestCodec.java
Modified: hadoop/common/trunk/CHANGES.txt
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/CHANGES.txt?rev=814455&r1=814454&r2=814455&view=diff
==============================================================================
--- hadoop/common/trunk/CHANGES.txt (original)
+++ hadoop/common/trunk/CHANGES.txt Mon Sep 14 00:45:01 2009
@@ -553,6 +553,9 @@
HADOOP-6252. Provide a method to determine if a deprecated key is set in
config file. (Jakob Homan via suresh)
+ HADOOP-5879. Read compression level and strategy from Configuration for
+ gzip compression. (He Yongqiang via cdouglas)
+
OPTIMIZATIONS
HADOOP-5595. NameNode does not need to run a replicator to choose a
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/CodecPool.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/CodecPool.java?rev=814455&r1=814454&r2=814455&view=diff
==============================================================================
--- hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/CodecPool.java
(original)
+++ hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/CodecPool.java
Mon Sep 14 00:45:01 2009
@@ -24,6 +24,7 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.ReflectionUtils;
/**
@@ -91,20 +92,26 @@
*
* @param codec the <code>CompressionCodec</code> for which to get the
* <code>Compressor</code>
+ * @param conf the <code>Configuration</code> object which contains confs
for creating or reinit the compressor
* @return <code>Compressor</code> for the given
* <code>CompressionCodec</code> from the pool or a new one
*/
- public static Compressor getCompressor(CompressionCodec codec) {
+ public static Compressor getCompressor(CompressionCodec codec, Configuration
conf) {
Compressor compressor = borrow(compressorPool, codec.getCompressorType());
if (compressor == null) {
compressor = codec.createCompressor();
LOG.info("Got brand-new compressor");
} else {
+ compressor.reinit(conf);
LOG.debug("Got recycled compressor");
}
return compressor;
}
+ public static Compressor getCompressor(CompressionCodec codec) {
+ return getCompressor(codec, null);
+ }
+
/**
* Get a {...@link Decompressor} for the given {...@link CompressionCodec}
from the
* pool or a new one.
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/Compressor.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/Compressor.java?rev=814455&r1=814454&r2=814455&view=diff
==============================================================================
--- hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/Compressor.java
(original)
+++ hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/Compressor.java
Mon Sep 14 00:45:01 2009
@@ -20,6 +20,8 @@
import java.io.IOException;
+import org.apache.hadoop.conf.Configuration;
+
/**
* Specification of a stream-based 'compressor' which can be
* plugged into a {...@link CompressionOutputStream} to compress data.
@@ -102,5 +104,13 @@
/**
* Closes the compressor and discards any unprocessed input.
*/
- public void end();
+ public void end();
+
+ /**
+ * Prepare the compressor to be used in a new stream with settings defined in
+ * the given Configuration
+ *
+ * @param conf Configuration from which new setting are fetched
+ */
+ public void reinit(Configuration conf);
}
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/GzipCodec.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/GzipCodec.java?rev=814455&r1=814454&r2=814455&view=diff
==============================================================================
--- hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/GzipCodec.java
(original)
+++ hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/GzipCodec.java
Mon Sep 14 00:45:01 2009
@@ -22,8 +22,11 @@
import java.util.zip.GZIPOutputStream;
import java.util.zip.GZIPInputStream;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.compress.zlib.*;
+import org.apache.hadoop.io.compress.zlib.ZlibCompressor.CompressionLevel;
+import org.apache.hadoop.io.compress.zlib.ZlibCompressor.CompressionStrategy;
/**
* This class creates gzip compressors/decompressors.
@@ -155,7 +158,7 @@
public Compressor createCompressor() {
return (ZlibFactory.isNativeZlibLoaded(conf))
- ? new GzipZlibCompressor()
+ ? new GzipZlibCompressor(conf)
: null;
}
@@ -206,6 +209,13 @@
ZlibCompressor.CompressionStrategy.DEFAULT_STRATEGY,
ZlibCompressor.CompressionHeader.GZIP_FORMAT, 64*1024);
}
+
+ public GzipZlibCompressor(Configuration conf) {
+ super(ZlibFactory.getCompressionLevel(conf),
+ ZlibFactory.getCompressionStrategy(conf),
+ ZlibCompressor.CompressionHeader.GZIP_FORMAT,
+ 64 * 1024);
+ }
}
static final class GzipZlibDecompressor extends ZlibDecompressor {
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/bzip2/BZip2DummyCompressor.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/bzip2/BZip2DummyCompressor.java?rev=814455&r1=814454&r2=814455&view=diff
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/bzip2/BZip2DummyCompressor.java
(original)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/bzip2/BZip2DummyCompressor.java
Mon Sep 14 00:45:01 2009
@@ -20,6 +20,7 @@
import java.io.IOException;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.compress.Compressor;
/**
@@ -77,4 +78,9 @@
throw new UnsupportedOperationException();
}
+ @Override
+ public void reinit(Configuration conf) {
+ // do nothing
+ }
+
}
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/zlib/BuiltInZlibDeflater.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/zlib/BuiltInZlibDeflater.java?rev=814455&r1=814454&r2=814455&view=diff
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/zlib/BuiltInZlibDeflater.java
(original)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/zlib/BuiltInZlibDeflater.java
Mon Sep 14 00:45:01 2009
@@ -21,7 +21,9 @@
import java.io.IOException;
import java.util.zip.Deflater;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.compress.Compressor;
+import org.mortbay.log.Log;
/**
* A wrapper around java.util.zip.Deflater to make it conform
@@ -46,4 +48,30 @@
throws IOException {
return super.deflate(b, off, len);
}
+
+ /**
+ * reinit the compressor with the given configuration. It will reset the
+ * compressor's compression level and compression strategy. Different from
+ * <tt>ZlibCompressor</tt>, <tt>BuiltInZlibDeflater</tt> only support three
+ * kind of compression strategy: FILTERED, HUFFMAN_ONLY and DEFAULT_STRATEGY.
+ * It will use DEFAULT_STRATEGY as default if the configured compression
+ * strategy is not supported.
+ */
+ @Override
+ public void reinit(Configuration conf) {
+ reset();
+ if (conf == null) {
+ return;
+ }
+ setLevel(ZlibFactory.getCompressionLevel(conf).compressionLevel());
+ final ZlibCompressor.CompressionStrategy strategy =
+ ZlibFactory.getCompressionStrategy(conf);
+ try {
+ setStrategy(strategy.compressionStrategy());
+ } catch (IllegalArgumentException ill) {
+ Log.warn(strategy + " not supported by BuiltInZlibDeflater.");
+ setStrategy(DEFAULT_STRATEGY);
+ }
+ Log.debug("Reinit compressor with new compression configuration");
+ }
}
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/zlib/ZlibCompressor.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/zlib/ZlibCompressor.java?rev=814455&r1=814454&r2=814455&view=diff
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/zlib/ZlibCompressor.java
(original)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/zlib/ZlibCompressor.java
Mon Sep 14 00:45:01 2009
@@ -22,8 +22,10 @@
import java.nio.Buffer;
import java.nio.ByteBuffer;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.compress.Compressor;
import org.apache.hadoop.util.NativeCodeLoader;
+import org.mortbay.log.Log;
/**
* A {...@link Compressor} based on the popular
@@ -40,7 +42,7 @@
private long stream;
private CompressionLevel level;
private CompressionStrategy strategy;
- private CompressionHeader windowBits;
+ private final CompressionHeader windowBits;
private int directBufferSize;
private byte[] userBuf = null;
private int userBufOff = 0, userBufLen = 0;
@@ -178,6 +180,31 @@
return nativeZlibLoaded;
}
+ protected final void construct(CompressionLevel level, CompressionStrategy
strategy,
+ CompressionHeader header, int directBufferSize) {
+ }
+
+ /**
+ * Creates a new compressor with the default compression level.
+ * Compressed data will be generated in ZLIB format.
+ */
+ public ZlibCompressor() {
+ this(CompressionLevel.DEFAULT_COMPRESSION,
+ CompressionStrategy.DEFAULT_STRATEGY,
+ CompressionHeader.DEFAULT_HEADER,
+ DEFAULT_DIRECT_BUFFER_SIZE);
+ }
+
+ /**
+ * Creates a new compressor, taking settings from the configuration.
+ */
+ public ZlibCompressor(Configuration conf) {
+ this(ZlibFactory.getCompressionLevel(conf),
+ ZlibFactory.getCompressionStrategy(conf),
+ CompressionHeader.DEFAULT_HEADER,
+ DEFAULT_DIRECT_BUFFER_SIZE);
+ }
+
/**
* Creates a new compressor using the specified compression level.
* Compressed data will be generated in ZLIB format.
@@ -192,28 +219,38 @@
this.level = level;
this.strategy = strategy;
this.windowBits = header;
+ stream = init(this.level.compressionLevel(),
+ this.strategy.compressionStrategy(),
+ this.windowBits.windowBits());
+
this.directBufferSize = directBufferSize;
-
uncompressedDirectBuf = ByteBuffer.allocateDirect(directBufferSize);
compressedDirectBuf = ByteBuffer.allocateDirect(directBufferSize);
compressedDirectBuf.position(directBufferSize);
-
- stream = init(this.level.compressionLevel(),
- this.strategy.compressionStrategy(),
- this.windowBits.windowBits());
}
-
+
/**
- * Creates a new compressor with the default compression level.
- * Compressed data will be generated in ZLIB format.
+ * Prepare the compressor to be used in a new stream with settings defined in
+ * the given Configuration. It will reset the compressor's compression level
+ * and compression strategy.
+ *
+ * @param conf Configuration storing new settings
*/
- public ZlibCompressor() {
- this(CompressionLevel.DEFAULT_COMPRESSION,
- CompressionStrategy.DEFAULT_STRATEGY,
- CompressionHeader.DEFAULT_HEADER,
- DEFAULT_DIRECT_BUFFER_SIZE);
+ @Override
+ public synchronized void reinit(Configuration conf) {
+ reset();
+ if (conf == null) {
+ return;
+ }
+ end(stream);
+ level = ZlibFactory.getCompressionLevel(conf);
+ strategy = ZlibFactory.getCompressionStrategy(conf);
+ stream = init(level.compressionLevel(),
+ strategy.compressionStrategy(),
+ windowBits.windowBits());
+ Log.debug("Reinit compressor with new compression configuration");
}
-
+
public synchronized void setInput(byte[] b, int off, int len) {
if (b== null) {
throw new NullPointerException();
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/zlib/ZlibFactory.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/zlib/ZlibFactory.java?rev=814455&r1=814454&r2=814455&view=diff
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/zlib/ZlibFactory.java
(original)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/compress/zlib/ZlibFactory.java
Mon Sep 14 00:45:01 2009
@@ -23,6 +23,8 @@
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.compress.Compressor;
import org.apache.hadoop.io.compress.Decompressor;
+import org.apache.hadoop.io.compress.zlib.ZlibCompressor.CompressionLevel;
+import org.apache.hadoop.io.compress.zlib.ZlibCompressor.CompressionStrategy;
import org.apache.hadoop.util.NativeCodeLoader;
/**
@@ -106,5 +108,25 @@
return (isNativeZlibLoaded(conf)) ?
new ZlibDecompressor() : new BuiltInZlibInflater();
}
-
+
+ public static void setCompressionStrategy(Configuration conf,
+ CompressionStrategy strategy) {
+ conf.setEnum("zlib.compress.strategy", strategy);
+ }
+
+ public static CompressionStrategy getCompressionStrategy(Configuration conf)
{
+ return conf.getEnum("zlib.compress.strategy",
+ CompressionStrategy.DEFAULT_STRATEGY);
+ }
+
+ public static void setCompressionLevel(Configuration conf,
+ CompressionLevel level) {
+ conf.setEnum("zlib.compress.level", level);
+ }
+
+ public static CompressionLevel getCompressionLevel(Configuration conf) {
+ return conf.getEnum("zlib.compress.level",
+ CompressionLevel.DEFAULT_COMPRESSION);
+ }
+
}
Modified:
hadoop/common/trunk/src/test/core/org/apache/hadoop/io/compress/TestCodec.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/test/core/org/apache/hadoop/io/compress/TestCodec.java?rev=814455&r1=814454&r2=814455&view=diff
==============================================================================
---
hadoop/common/trunk/src/test/core/org/apache/hadoop/io/compress/TestCodec.java
(original)
+++
hadoop/common/trunk/src/test/core/org/apache/hadoop/io/compress/TestCodec.java
Mon Sep 14 00:45:01 2009
@@ -20,9 +20,11 @@
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
+import java.util.Arrays;
import java.util.Random;
import org.apache.hadoop.conf.Configuration;
@@ -39,6 +41,8 @@
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.compress.CompressionOutputStream;
+import org.apache.hadoop.io.compress.zlib.ZlibCompressor.CompressionLevel;
+import org.apache.hadoop.io.compress.zlib.ZlibCompressor.CompressionStrategy;
import org.apache.hadoop.io.compress.zlib.ZlibFactory;
import org.apache.hadoop.util.LineReader;
import org.apache.hadoop.util.ReflectionUtils;
@@ -77,6 +81,15 @@
codecTest(conf, seed, count, "org.apache.hadoop.io.compress.BZip2Codec");
}
+ @Test
+ public void testGzipCodecWithParam() throws IOException {
+ Configuration conf = new Configuration(this.conf);
+ ZlibFactory.setCompressionLevel(conf, CompressionLevel.BEST_COMPRESSION);
+ ZlibFactory.setCompressionStrategy(conf, CompressionStrategy.HUFFMAN_ONLY);
+ codecTest(conf, seed, 0, "org.apache.hadoop.io.compress.GzipCodec");
+ codecTest(conf, seed, count, "org.apache.hadoop.io.compress.GzipCodec");
+ }
+
private static void codecTest(Configuration conf, int seed, int count,
String codecClass)
throws IOException {
@@ -262,6 +275,54 @@
assertTrue("Got mismatched ZlibCompressor", c2 !=
CodecPool.getCompressor(gzc));
}
+ private static void gzipReinitTest(Configuration conf, CompressionCodec
codec)
+ throws IOException {
+ // Add codec to cache
+ ZlibFactory.setCompressionLevel(conf, CompressionLevel.BEST_COMPRESSION);
+ ZlibFactory.setCompressionStrategy(conf,
+ CompressionStrategy.DEFAULT_STRATEGY);
+ Compressor c1 = CodecPool.getCompressor(codec);
+ CodecPool.returnCompressor(c1);
+ // reset compressor's compression level to perform no compression
+ ZlibFactory.setCompressionLevel(conf, CompressionLevel.NO_COMPRESSION);
+ Compressor c2 = CodecPool.getCompressor(codec, conf);
+ // ensure same compressor placed earlier
+ assertTrue("Got mismatched ZlibCompressor", c1 == c2);
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ CompressionOutputStream cos = null;
+ // write trivially compressable data
+ byte[] b = new byte[1 << 15];
+ Arrays.fill(b, (byte) 43);
+ try {
+ cos = codec.createOutputStream(bos, c2);
+ cos.write(b);
+ } finally {
+ if (cos != null) {
+ cos.close();
+ }
+ CodecPool.returnCompressor(c2);
+ }
+ byte[] outbytes = bos.toByteArray();
+ // verify data were not compressed
+ assertTrue("Compressed bytes contrary to configuration",
+ outbytes.length >= b.length);
+ }
+
+ @Test
+ public void testCodecPoolCompressorReinit() throws Exception {
+ Configuration conf = new Configuration();
+ conf.setBoolean("hadoop.native.lib", true);
+ if (ZlibFactory.isNativeZlibLoaded(conf)) {
+ GzipCodec gzc = ReflectionUtils.newInstance(GzipCodec.class, conf);
+ gzipReinitTest(conf, gzc);
+ } else {
+ LOG.warn("testCodecPoolCompressorReinit skipped: native libs not
loaded");
+ }
+ conf.setBoolean("hadoop.native.lib", false);
+ DefaultCodec dfc = ReflectionUtils.newInstance(DefaultCodec.class, conf);
+ gzipReinitTest(conf, dfc);
+ }
+
@Test
public void testSequenceFileDefaultCodec() throws IOException,
ClassNotFoundException,
InstantiationException, IllegalAccessException {
