textinputformat.java

来自「Hadoop是一个用于运行应用程序在大型集群的廉价硬件设备上的框架。Hadoop」· Java 代码 · 共 97 行

/**
 * Copyright 2005 The Apache Software Foundation
 *
 * Licensed 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.mapred;

import java.io.IOException;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FSDataInputStream;

import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.UTF8;

/** An {@link InputFormat} for plain text files.  Files are broken into lines.
 * Either linefeed or carriage-return are used to signal end of line.  Keys are
 * the position in the file, and values are the line of text.. */
public class TextInputFormat extends InputFormatBase {

  public RecordReader getRecordReader(FileSystem fs, FileSplit split,
                                      JobConf job, Reporter reporter)
    throws IOException {

    reporter.setStatus(split.toString());

    final long start = split.getStart();
    final long end = start + split.getLength();

    // open the file and seek to the start of the split
    final FSDataInputStream in = fs.open(split.getFile());
    
    if (start != 0) {
      in.seek(start-1);
      while (in.getPos() < end) {    // scan to the next newline in the file
        char c = (char)in.read();
        if (c == '\r' || c == '\n') {
          break;
        }
      }
    }

    return new RecordReader() {
        /** Read a line. */
        public synchronized boolean next(Writable key, Writable value)
          throws IOException {
          long pos = in.getPos();
          if (pos >= end)
            return false;

          ((LongWritable)key).set(pos);           // key is position
          ((UTF8)value).set(readLine(in));        // value is line
          return true;
        }
        
        public  synchronized long getPos() throws IOException {
          return in.getPos();
        }

        public synchronized void close() throws IOException { in.close(); }

      };
  }

  private static String readLine(FSDataInputStream in) throws IOException {
    StringBuffer buffer = new StringBuffer();
    while (true) {

      int b = in.read();
      if (b == -1)
        break;

      char c = (char)b;              // bug: this assumes eight-bit characters.
      if (c == '\r' || c == '\n')
        break;

      buffer.append(c);
    }
    
    return buffer.toString();
  }

}