Author: shv
Date: Tue Sep 25 21:25:56 2012
New Revision: 1390150
URL: http://svn.apache.org/viewvc?rev=1390150&view=rev
Log:
MAPREDUCE-2786. Add compression option for TestDFSIO. Contributed by Plamen
Jeliazkov.
Modified:
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/fs/IOMapperBase.java
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/fs/TestDFSIO.java
Modified:
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt?rev=1390150&r1=1390149&r2=1390150&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt
(original)
+++ hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt Tue
Sep 25 21:25:56 2012
@@ -8,6 +8,9 @@ Release 0.23.4 - UNRELEASED
IMPROVEMENTS
+ MAPREDUCE-2786. Add compression option for TestDFSIO.
+ (Plamen Jeliazkov via shv)
+
MAPREDUCE-4645. Provide a random seed to Slive to make the sequence
of file names deterministic. (Ravi Prakash via shv)
Modified:
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/fs/IOMapperBase.java
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/fs/IOMapperBase.java?rev=1390150&r1=1390149&r2=1390150&view=diff
==============================================================================
---
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/fs/IOMapperBase.java
(original)
+++
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/fs/IOMapperBase.java
Tue Sep 25 21:25:56 2012
@@ -22,7 +22,9 @@ import java.net.InetAddress;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.mapred.*;
+import org.apache.hadoop.util.ReflectionUtils;
/**
* Base mapper class for IO operations.
@@ -41,6 +43,7 @@ public abstract class IOMapperBase<T> ex
protected int bufferSize;
protected FileSystem fs;
protected String hostName;
+ protected CompressionCodec compressionCodec;
public IOMapperBase() {
}
@@ -59,6 +62,22 @@ public abstract class IOMapperBase<T> ex
} catch(Exception e) {
hostName = "localhost";
}
+
+ //grab compression
+ String compression = getConf().get("test.io.compression.class", null);
+ Class<? extends CompressionCodec> codec;
+
+ //try to initialize codec
+ try {
+ codec = (compression == null) ? null :
+ Class.forName(compression).asSubclass(CompressionCodec.class);
+ } catch(Exception e) {
+ throw new RuntimeException("Compression codec not found: ", e);
+ }
+
+ if(codec != null) {
+ compressionCodec = (CompressionCodec) ReflectionUtils.newInstance(codec,
getConf());
+ }
}
public void close() throws IOException {
Modified:
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/fs/TestDFSIO.java
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/fs/TestDFSIO.java?rev=1390150&r1=1390149&r2=1390150&view=diff
==============================================================================
---
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/fs/TestDFSIO.java
(original)
+++
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/fs/TestDFSIO.java
Tue Sep 25 21:25:56 2012
@@ -23,6 +23,7 @@ import java.io.DataInputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
+import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PrintStream;
@@ -295,6 +296,8 @@ public class TestDFSIO extends TestCase
// create file
OutputStream out;
out = fs.create(new Path(getDataDir(getConf()), name), true, bufferSize);
+
+ if(compressionCodec != null) out =
compressionCodec.createOutputStream(out);
try {
// write to the file
@@ -358,6 +361,8 @@ public class TestDFSIO extends TestCase
OutputStream out;
out = fs.append(new Path(getDataDir(getConf()), name), bufferSize);
+ if(compressionCodec != null) out =
compressionCodec.createOutputStream(out);
+
try {
// write to the file
long nrRemaining;
@@ -394,7 +399,10 @@ public class TestDFSIO extends TestCase
long totalSize // in bytes
) throws IOException {
// open file
- DataInputStream in = fs.open(new Path(getDataDir(getConf()), name));
+ InputStream in = fs.open(new Path(getDataDir(getConf()), name));
+
+ if(compressionCodec != null) in = compressionCodec.createInputStream(in);
+
long actualSize = 0;
try {
while (actualSize < totalSize) {
@@ -459,6 +467,7 @@ public class TestDFSIO extends TestCase
long fileSize = 1*MEGA;
int nrFiles = 1;
String resFileName = DEFAULT_RES_FILE_NAME;
+ String compressionClass = null;
boolean isSequential = false;
String version = TestDFSIO.class.getSimpleName() + ".0.0.6";
@@ -479,6 +488,8 @@ public class TestDFSIO extends TestCase
testType = TEST_TYPE_CLEANUP;
} else if (args[i].startsWith("-seq")) {
isSequential = true;
+ } else if (args[i].startsWith("-compression")) {
+ compressionClass = args[++i];
} else if (args[i].equals("-nrFiles")) {
nrFiles = Integer.parseInt(args[++i]);
} else if (args[i].equals("-fileSize")) {
@@ -497,6 +508,11 @@ public class TestDFSIO extends TestCase
LOG.info("fileSize (MB) = " + toMB(fileSize));
LOG.info("bufferSize = " + bufferSize);
LOG.info("baseDir = " + getBaseDir(config));
+
+ if(compressionClass != null) {
+ config.set("test.io.compression.class", compressionClass);
+ LOG.info("compressionClass = " + compressionClass);
+ }
config.setInt("test.io.file.buffer.size", bufferSize);
config.setBoolean(DFSConfigKeys.DFS_SUPPORT_APPEND_KEY, true);
