You are viewing a plain text version of this content. The canonical link for it is here.
Posted to mapreduce-commits@hadoop.apache.org by to...@apache.org on 2011/02/24 08:07:30 UTC
svn commit: r1074055 - in /hadoop/mapreduce/trunk: ./
src/java/org/apache/hadoop/mapreduce/lib/input/
src/test/mapred/org/apache/hadoop/mapreduce/lib/input/
Author: todd
Date: Thu Feb 24 07:07:29 2011
New Revision: 1074055
URL: http://svn.apache.org/viewvc?rev=1074055&view=rev
Log:
MAPREDUCE-2254. Allow setting of end-of-record delimiter for TextInputFormat. Contributed by Ahmed Radwan.
Added:
hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/lib/input/TestLineRecordReader.java
Modified:
hadoop/mapreduce/trunk/CHANGES.txt
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/input/LineRecordReader.java
hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/input/TextInputFormat.java
Modified: hadoop/mapreduce/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/CHANGES.txt?rev=1074055&r1=1074054&r2=1074055&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/CHANGES.txt (original)
+++ hadoop/mapreduce/trunk/CHANGES.txt Thu Feb 24 07:07:29 2011
@@ -32,6 +32,9 @@ Trunk (unreleased changes)
MAPREDUCE-2334. Update BlockPlacementPolicyRaid for the new method
in BlockPlacementPolicy. (szetszwo)
+ MAPREDUCE-2254. Allow setting of end-of-record delimiter for
+ TextInputFormat (Ahmed Radwan via todd)
+
OPTIMIZATIONS
MAPREDUCE-2026. Make JobTracker.getJobCounters() and
Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/input/LineRecordReader.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/input/LineRecordReader.java?rev=1074055&r1=1074054&r2=1074055&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/input/LineRecordReader.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/input/LineRecordReader.java Thu Feb 24 07:07:29 2011
@@ -66,6 +66,14 @@ public class LineRecordReader extends Re
private Counter inputByteCounter;
private CompressionCodec codec;
private Decompressor decompressor;
+ private byte[] recordDelimiterBytes;
+
+ public LineRecordReader() {
+ }
+
+ public LineRecordReader(byte[] recordDelimiter) {
+ this.recordDelimiterBytes = recordDelimiter;
+ }
public void initialize(InputSplit genericSplit,
TaskAttemptContext context) throws IOException {
@@ -90,17 +98,33 @@ public class LineRecordReader extends Re
((SplittableCompressionCodec)codec).createInputStream(
fileIn, decompressor, start, end,
SplittableCompressionCodec.READ_MODE.BYBLOCK);
- in = new LineReader(cIn, job);
+ if (null == this.recordDelimiterBytes){
+ in = new LineReader(cIn, job);
+ } else {
+ in = new LineReader(cIn, job, this.recordDelimiterBytes);
+ }
+
start = cIn.getAdjustedStart();
end = cIn.getAdjustedEnd();
filePosition = cIn;
} else {
- in = new LineReader(codec.createInputStream(fileIn, decompressor), job);
+ if (null == this.recordDelimiterBytes) {
+ in = new LineReader(codec.createInputStream(fileIn, decompressor),
+ job);
+ } else {
+ in = new LineReader(codec.createInputStream(fileIn,
+ decompressor), job, this.recordDelimiterBytes);
+ }
filePosition = fileIn;
}
} else {
fileIn.seek(start);
- in = new LineReader(fileIn, job);
+ if (null == this.recordDelimiterBytes){
+ in = new LineReader(fileIn, job);
+ } else {
+ in = new LineReader(fileIn, job, this.recordDelimiterBytes);
+ }
+
filePosition = fileIn;
}
// If this is not the first split, we always throw away first record
Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/input/TextInputFormat.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/input/TextInputFormat.java?rev=1074055&r1=1074054&r2=1074055&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/input/TextInputFormat.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/input/TextInputFormat.java Thu Feb 24 07:07:29 2011
@@ -43,7 +43,12 @@ public class TextInputFormat extends Fil
public RecordReader<LongWritable, Text>
createRecordReader(InputSplit split,
TaskAttemptContext context) {
- return new LineRecordReader();
+ String delimiter = context.getConfiguration().get(
+ "textinputformat.record.delimiter");
+ byte[] recordDelimiterBytes = null;
+ if (null != delimiter)
+ recordDelimiterBytes = delimiter.getBytes();
+ return new LineRecordReader(recordDelimiterBytes);
}
@Override
Added: hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/lib/input/TestLineRecordReader.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/lib/input/TestLineRecordReader.java?rev=1074055&view=auto
==============================================================================
--- hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/lib/input/TestLineRecordReader.java (added)
+++ hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/lib/input/TestLineRecordReader.java Thu Feb 24 07:07:29 2011
@@ -0,0 +1,140 @@
+/**
+ * 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.mapreduce.lib.input;
+
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
+import java.io.Reader;
+import java.io.Writer;
+
+import junit.framework.TestCase;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.tools.ant.util.FileUtils;
+import org.junit.Test;
+
+public class TestLineRecordReader extends TestCase {
+
+ private static Path workDir = new Path(new Path(System.getProperty(
+ "test.build.data", "."), "data"), "TestTextInputFormat");
+ private static Path inputDir = new Path(workDir, "input");
+ private static Path outputDir = new Path(workDir, "output");
+
+ /**
+ * Writes the input test file
+ *
+ * @param conf
+ * @throws IOException
+ */
+ public void createInputFile(Configuration conf) throws IOException {
+ FileSystem localFs = FileSystem.getLocal(conf);
+ Path file = new Path(inputDir, "test.txt");
+ Writer writer = new OutputStreamWriter(localFs.create(file));
+ writer.write("abc\ndef\t\nghi\njkl");
+ writer.close();
+ }
+
+ /**
+ * Reads the output file into a string
+ *
+ * @param conf
+ * @return
+ * @throws IOException
+ */
+ public String readOutputFile(Configuration conf) throws IOException {
+ FileSystem localFs = FileSystem.getLocal(conf);
+ Path file = new Path(outputDir, "part-r-00000");
+ Reader reader = new InputStreamReader(localFs.open(file));
+ String r = FileUtils.readFully(reader);
+ reader.close();
+ return r;
+ }
+
+ /**
+ * Creates and runs an MR job
+ *
+ * @param conf
+ * @throws IOException
+ * @throws InterruptedException
+ * @throws ClassNotFoundException
+ */
+ public void createAndRunJob(Configuration conf) throws IOException,
+ InterruptedException, ClassNotFoundException {
+ Job job = Job.getInstance(conf);
+ job.setJarByClass(TestLineRecordReader.class);
+ job.setMapperClass(Mapper.class);
+ job.setReducerClass(Reducer.class);
+ FileInputFormat.addInputPath(job, inputDir);
+ FileOutputFormat.setOutputPath(job, outputDir);
+ job.waitForCompletion(true);
+ }
+
+ /**
+ * Test the case when a custom record delimiter is specified using the
+ * textinputformat.record.delimiter configuration property
+ *
+ * @throws IOException
+ * @throws InterruptedException
+ * @throws ClassNotFoundException
+ */
+ @Test
+ public void testCustomRecordDelimiters() throws IOException,
+ InterruptedException, ClassNotFoundException {
+ Configuration conf = new Configuration();
+ conf.set("textinputformat.record.delimiter", "\t\n");
+ FileSystem localFs = FileSystem.getLocal(conf);
+ // cleanup
+ localFs.delete(workDir, true);
+ // creating input test file
+ createInputFile(conf);
+ createAndRunJob(conf);
+ String expected = "0\tabc\ndef\n9\tghi\njkl\n";
+ this.assertEquals(expected, readOutputFile(conf));
+ }
+
+ /**
+ * Test the default behavior when the textinputformat.record.delimiter
+ * configuration property is not specified
+ *
+ * @throws IOException
+ * @throws InterruptedException
+ * @throws ClassNotFoundException
+ */
+ @Test
+ public void testDefaultRecordDelimiters() throws IOException,
+ InterruptedException, ClassNotFoundException {
+ Configuration conf = new Configuration();
+ FileSystem localFs = FileSystem.getLocal(conf);
+ // cleanup
+ localFs.delete(workDir, true);
+ // creating input test file
+ createInputFile(conf);
+ createAndRunJob(conf);
+ String expected = "0\tabc\n4\tdef\t\n9\tghi\n13\tjkl\n";
+ this.assertEquals(expected, readOutputFile(conf));
+ }
+
+}