001 /**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements. See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership. The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License. You may obtain a copy of the License at
009 *
010 * http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018
019 package org.apache.hadoop.mapred;
020
021 import java.io.*;
022
023 import org.apache.hadoop.classification.InterfaceAudience;
024 import org.apache.hadoop.classification.InterfaceStability;
025 import org.apache.hadoop.fs.*;
026 import org.apache.hadoop.io.LongWritable;
027 import org.apache.hadoop.io.Text;
028 import org.apache.hadoop.io.compress.*;
029
030 /**
031 * An {@link InputFormat} for plain text files. Files are broken into lines.
032 * Either linefeed or carriage-return are used to signal end of line. Keys are
033 * the position in the file, and values are the line of text..
034 */
035 @InterfaceAudience.Public
036 @InterfaceStability.Stable
037 public class TextInputFormat extends FileInputFormat<LongWritable, Text>
038 implements JobConfigurable {
039
040 private CompressionCodecFactory compressionCodecs = null;
041
042 public void configure(JobConf conf) {
043 compressionCodecs = new CompressionCodecFactory(conf);
044 }
045
046 protected boolean isSplitable(FileSystem fs, Path file) {
047 final CompressionCodec codec = compressionCodecs.getCodec(file);
048 if (null == codec) {
049 return true;
050 }
051 return codec instanceof SplittableCompressionCodec;
052 }
053
054 public RecordReader<LongWritable, Text> getRecordReader(
055 InputSplit genericSplit, JobConf job,
056 Reporter reporter)
057 throws IOException {
058
059 reporter.setStatus(genericSplit.toString());
060 String delimiter = job.get("textinputformat.record.delimiter");
061 byte[] recordDelimiterBytes = null;
062 if (null != delimiter) recordDelimiterBytes = delimiter.getBytes();
063 return new LineRecordReader(job, (FileSplit) genericSplit,
064 recordDelimiterBytes);
065 }
066 }