You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by sa...@apache.org on 2013/10/18 23:03:52 UTC
svn commit: r1533631 - in /hadoop/common/branches/branch-1: ./
src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/
src/contrib/streaming/src/test/org/apache/hadoop/streaming/
src/contrib/streaming/src/test/org/apache/hadoop/streaming/io/
Author: sandy
Date: Fri Oct 18 21:03:52 2013
New Revision: 1533631
URL: http://svn.apache.org/r1533631
Log:
MAPREDUCE-5457. Add a KeyOnlyTextOutputReader to enable streaming to write out text files without separators (Sandy Ryza)
Added:
hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/KeyOnlyTextInputWriter.java
hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/KeyOnlyTextOutputReader.java
hadoop/common/branches/branch-1/src/contrib/streaming/src/test/org/apache/hadoop/streaming/TestStreamingOutputOnlyKeys.java
hadoop/common/branches/branch-1/src/contrib/streaming/src/test/org/apache/hadoop/streaming/io/
hadoop/common/branches/branch-1/src/contrib/streaming/src/test/org/apache/hadoop/streaming/io/TestKeyOnlyTextOutputReader.java
Modified:
hadoop/common/branches/branch-1/CHANGES.txt
hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/IdentifierResolver.java
hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/TextInputWriter.java
Modified: hadoop/common/branches/branch-1/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/CHANGES.txt?rev=1533631&r1=1533630&r2=1533631&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/CHANGES.txt (original)
+++ hadoop/common/branches/branch-1/CHANGES.txt Fri Oct 18 21:03:52 2013
@@ -160,6 +160,9 @@ Release 1.3.0 - unreleased
HDFS-4794. Browsing filesystem via webui throws kerberos exception when NN
service RPC is enabled in a secure cluster. (Benoy Antony via jing9)
+ MAPRDUCE-5457. Add a KeyOnlyTextOutputReader to enable streaming to write
+ out text files without separators (Sandy Ryza)
+
Release 1.2.2 - unreleased
INCOMPATIBLE CHANGES
Modified: hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/IdentifierResolver.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/IdentifierResolver.java?rev=1533631&r1=1533630&r2=1533631&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/IdentifierResolver.java (original)
+++ hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/IdentifierResolver.java Fri Oct 18 21:03:52 2013
@@ -19,6 +19,7 @@
package org.apache.hadoop.streaming.io;
import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.typedbytes.TypedBytesWritable;
@@ -34,7 +35,8 @@ public class IdentifierResolver {
public static final String TEXT_ID = "text";
public static final String RAW_BYTES_ID = "rawbytes";
public static final String TYPED_BYTES_ID = "typedbytes";
-
+ public static final String KEY_ONLY_TEXT_ID = "keyonlytext";
+
private Class<? extends InputWriter> inputWriterClass = null;
private Class<? extends OutputReader> outputReaderClass = null;
private Class outputKeyClass = null;
@@ -55,6 +57,11 @@ public class IdentifierResolver {
setOutputReaderClass(TypedBytesOutputReader.class);
setOutputKeyClass(TypedBytesWritable.class);
setOutputValueClass(TypedBytesWritable.class);
+ } else if (identifier.equalsIgnoreCase(KEY_ONLY_TEXT_ID)) {
+ setInputWriterClass(KeyOnlyTextInputWriter.class);
+ setOutputReaderClass(KeyOnlyTextOutputReader.class);
+ setOutputKeyClass(Text.class);
+ setOutputValueClass(NullWritable.class);
} else { // assume TEXT_ID
setInputWriterClass(TextInputWriter.class);
setOutputReaderClass(TextOutputReader.class);
Added: hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/KeyOnlyTextInputWriter.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/KeyOnlyTextInputWriter.java?rev=1533631&view=auto
==============================================================================
--- hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/KeyOnlyTextInputWriter.java (added)
+++ hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/KeyOnlyTextInputWriter.java Fri Oct 18 21:03:52 2013
@@ -0,0 +1,35 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.streaming.io;
+
+import java.io.IOException;
+
+
+public class KeyOnlyTextInputWriter extends TextInputWriter {
+
+ @Override
+ public void writeKey(Object key) throws IOException {
+ writeUTF8(key);
+ clientOut.write('\n');
+ }
+
+ @Override
+ public void writeValue(Object value) throws IOException {}
+
+}
Added: hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/KeyOnlyTextOutputReader.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/KeyOnlyTextOutputReader.java?rev=1533631&view=auto
==============================================================================
--- hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/KeyOnlyTextOutputReader.java (added)
+++ hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/KeyOnlyTextOutputReader.java Fri Oct 18 21:03:52 2013
@@ -0,0 +1,90 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.streaming.io;
+
+import java.io.DataInput;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UnsupportedEncodingException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.streaming.PipeMapRed;
+import org.apache.hadoop.util.LineReader;
+
+/**
+ * OutputReader that reads the client's output as text, interpreting each line
+ * as a key and outputting NullWritables for values.
+ */
+public class KeyOnlyTextOutputReader extends OutputReader<Text, NullWritable> {
+
+ private LineReader lineReader;
+ private byte[] bytes;
+ private DataInput clientIn;
+ private Configuration conf;
+ private Text key;
+ private Text line;
+
+ @Override
+ public void initialize(PipeMapRed pipeMapRed) throws IOException {
+ super.initialize(pipeMapRed);
+ clientIn = pipeMapRed.getClientInput();
+ conf = pipeMapRed.getConfiguration();
+ lineReader = new LineReader((InputStream)clientIn, conf);
+ key = new Text();
+ line = new Text();
+ }
+
+ @Override
+ public boolean readKeyValue() throws IOException {
+ if (lineReader.readLine(line) <= 0) {
+ return false;
+ }
+ bytes = line.getBytes();
+ key.set(bytes, 0, line.getLength());
+
+ line.clear();
+ return true;
+ }
+
+ @Override
+ public Text getCurrentKey() throws IOException {
+ return key;
+ }
+
+ @Override
+ public NullWritable getCurrentValue() throws IOException {
+ return NullWritable.get();
+ }
+
+ @Override
+ public String getLastOutput() {
+ if (bytes != null) {
+ try {
+ return new String(bytes, "UTF-8");
+ } catch (UnsupportedEncodingException e) {
+ return "<undecodable>";
+ }
+ } else {
+ return null;
+ }
+ }
+
+}
Modified: hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/TextInputWriter.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/TextInputWriter.java?rev=1533631&r1=1533630&r2=1533631&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/TextInputWriter.java (original)
+++ hadoop/common/branches/branch-1/src/contrib/streaming/src/java/org/apache/hadoop/streaming/io/TextInputWriter.java Fri Oct 18 21:03:52 2013
@@ -30,7 +30,7 @@ import org.apache.hadoop.streaming.PipeM
*/
public class TextInputWriter extends InputWriter<Object, Object> {
- private DataOutput clientOut;
+ protected DataOutput clientOut;
private byte[] inputSeparator;
@Override
@@ -53,7 +53,7 @@ public class TextInputWriter extends Inp
}
// Write an object to the output stream using UTF-8 encoding
- private void writeUTF8(Object object) throws IOException {
+ protected void writeUTF8(Object object) throws IOException {
byte[] bval;
int valSize;
if (object instanceof BytesWritable) {
Added: hadoop/common/branches/branch-1/src/contrib/streaming/src/test/org/apache/hadoop/streaming/TestStreamingOutputOnlyKeys.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/contrib/streaming/src/test/org/apache/hadoop/streaming/TestStreamingOutputOnlyKeys.java?rev=1533631&view=auto
==============================================================================
--- hadoop/common/branches/branch-1/src/contrib/streaming/src/test/org/apache/hadoop/streaming/TestStreamingOutputOnlyKeys.java (added)
+++ hadoop/common/branches/branch-1/src/contrib/streaming/src/test/org/apache/hadoop/streaming/TestStreamingOutputOnlyKeys.java Fri Oct 18 21:03:52 2013
@@ -0,0 +1,50 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.streaming;
+
+import java.io.IOException;
+
+public class TestStreamingOutputOnlyKeys extends TestStreaming {
+
+ public TestStreamingOutputOnlyKeys() throws IOException {
+ super();
+ outputExpect = outputExpect.replaceAll("\t", "");
+ }
+
+ protected String[] genArgs() {
+ return new String[] {
+ "-input", INPUT_FILE.getAbsolutePath(),
+ "-output", OUTPUT_DIR.getAbsolutePath(),
+ "-mapper", map,
+ "-reducer", reduce,
+ //"-verbose",
+ //"-jobconf", "stream.debug=set"
+ "-jobconf", "stream.reduce.input=keyonlytext",
+ "-jobconf", "stream.reduce.output=keyonlytext",
+ "-jobconf", "keep.failed.task.files=true",
+ "-jobconf", "stream.tmpdir="+System.getProperty("test.build.data","/tmp")
+ };
+ }
+
+ @Override
+ public void testCommandLine() {
+ // Do nothing
+ }
+
+}
Added: hadoop/common/branches/branch-1/src/contrib/streaming/src/test/org/apache/hadoop/streaming/io/TestKeyOnlyTextOutputReader.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/contrib/streaming/src/test/org/apache/hadoop/streaming/io/TestKeyOnlyTextOutputReader.java?rev=1533631&view=auto
==============================================================================
--- hadoop/common/branches/branch-1/src/contrib/streaming/src/test/org/apache/hadoop/streaming/io/TestKeyOnlyTextOutputReader.java (added)
+++ hadoop/common/branches/branch-1/src/contrib/streaming/src/test/org/apache/hadoop/streaming/io/TestKeyOnlyTextOutputReader.java Fri Oct 18 21:03:52 2013
@@ -0,0 +1,68 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.streaming.io;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.IOException;
+
+import junit.framework.Assert;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.streaming.PipeMapRed;
+import org.apache.hadoop.streaming.PipeMapper;
+import org.junit.Test;
+
+public class TestKeyOnlyTextOutputReader {
+ @Test
+ public void testKeyOnlyTextOutputReader() throws IOException {
+ String text = "key,value\nkey2,value2\nnocomma\n";
+ PipeMapRed pipeMapRed = new MyPipeMapRed(text);
+ KeyOnlyTextOutputReader outputReader = new KeyOnlyTextOutputReader();
+ outputReader.initialize(pipeMapRed);
+ outputReader.readKeyValue();
+ Assert.assertEquals(new Text("key,value"), outputReader.getCurrentKey());
+ outputReader.readKeyValue();
+ Assert.assertEquals(new Text("key2,value2"), outputReader.getCurrentKey());
+ outputReader.readKeyValue();
+ Assert.assertEquals(new Text("nocomma"), outputReader.getCurrentKey());
+ Assert.assertEquals(false, outputReader.readKeyValue());
+ }
+
+ private class MyPipeMapRed extends PipeMapper {
+ private DataInput clientIn;
+ private Configuration conf = new Configuration();
+
+ public MyPipeMapRed(String text) {
+ clientIn = new DataInputStream(new ByteArrayInputStream(text.getBytes()));
+ }
+
+ @Override
+ public DataInput getClientInput() {
+ return clientIn;
+ }
+
+ @Override
+ public Configuration getConfiguration() {
+ return conf;
+ }
+ }
+}