LineRecordReader.java
上传用户:quxuerui
上传日期:2018-01-08
资源大小:41811k
文件大小:5k
源码类别:

网格计算

开发平台:

Java

  1. /**
  2.  * Licensed to the Apache Software Foundation (ASF) under one
  3.  * or more contributor license agreements.  See the NOTICE file
  4.  * distributed with this work for additional information
  5.  * regarding copyright ownership.  The ASF licenses this file
  6.  * to you under the Apache License, Version 2.0 (the
  7.  * "License"); you may not use this file except in compliance
  8.  * with the License.  You may obtain a copy of the License at
  9.  *
  10.  *     http://www.apache.org/licenses/LICENSE-2.0
  11.  *
  12.  * Unless required by applicable law or agreed to in writing, software
  13.  * distributed under the License is distributed on an "AS IS" BASIS,
  14.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15.  * See the License for the specific language governing permissions and
  16.  * limitations under the License.
  17.  */
  18. package org.apache.hadoop.mapred;
  19. import java.io.IOException;
  20. import java.io.InputStream;
  21. import org.apache.hadoop.conf.Configuration;
  22. import org.apache.hadoop.fs.FSDataInputStream;
  23. import org.apache.hadoop.fs.FileSystem;
  24. import org.apache.hadoop.fs.Path;
  25. import org.apache.hadoop.io.LongWritable;
  26. import org.apache.hadoop.io.Text;
  27. import org.apache.hadoop.io.compress.CompressionCodec;
  28. import org.apache.hadoop.io.compress.CompressionCodecFactory;
  29. import org.apache.commons.logging.LogFactory;
  30. import org.apache.commons.logging.Log;
  31. /**
  32.  * Treats keys as offset in file and value as line. 
  33.  * @deprecated Use 
  34.  *   {@link org.apache.hadoop.mapreduce.lib.input.LineRecordReader} instead.
  35.  */
  36. @Deprecated
  37. public class LineRecordReader implements RecordReader<LongWritable, Text> {
  38.   private static final Log LOG
  39.     = LogFactory.getLog(LineRecordReader.class.getName());
  40.   private CompressionCodecFactory compressionCodecs = null;
  41.   private long start;
  42.   private long pos;
  43.   private long end;
  44.   private LineReader in;
  45.   int maxLineLength;
  46.   /**
  47.    * A class that provides a line reader from an input stream.
  48.    * @deprecated Use {@link org.apache.hadoop.util.LineReader} instead.
  49.    */
  50.   @Deprecated
  51.   public static class LineReader extends org.apache.hadoop.util.LineReader {
  52.     LineReader(InputStream in) {
  53.       super(in);
  54.     }
  55.     LineReader(InputStream in, int bufferSize) {
  56.       super(in, bufferSize);
  57.     }
  58.     public LineReader(InputStream in, Configuration conf) throws IOException {
  59.       super(in, conf);
  60.     }
  61.   }
  62.   public LineRecordReader(Configuration job, 
  63.                           FileSplit split) throws IOException {
  64.     this.maxLineLength = job.getInt("mapred.linerecordreader.maxlength",
  65.                                     Integer.MAX_VALUE);
  66.     start = split.getStart();
  67.     end = start + split.getLength();
  68.     final Path file = split.getPath();
  69.     compressionCodecs = new CompressionCodecFactory(job);
  70.     final CompressionCodec codec = compressionCodecs.getCodec(file);
  71.     // open the file and seek to the start of the split
  72.     FileSystem fs = file.getFileSystem(job);
  73.     FSDataInputStream fileIn = fs.open(split.getPath());
  74.     boolean skipFirstLine = false;
  75.     if (codec != null) {
  76.       in = new LineReader(codec.createInputStream(fileIn), job);
  77.       end = Long.MAX_VALUE;
  78.     } else {
  79.       if (start != 0) {
  80.         skipFirstLine = true;
  81.         --start;
  82.         fileIn.seek(start);
  83.       }
  84.       in = new LineReader(fileIn, job);
  85.     }
  86.     if (skipFirstLine) {  // skip first line and re-establish "start".
  87.       start += in.readLine(new Text(), 0,
  88.                            (int)Math.min((long)Integer.MAX_VALUE, end - start));
  89.     }
  90.     this.pos = start;
  91.   }
  92.   
  93.   public LineRecordReader(InputStream in, long offset, long endOffset,
  94.                           int maxLineLength) {
  95.     this.maxLineLength = maxLineLength;
  96.     this.in = new LineReader(in);
  97.     this.start = offset;
  98.     this.pos = offset;
  99.     this.end = endOffset;    
  100.   }
  101.   public LineRecordReader(InputStream in, long offset, long endOffset, 
  102.                           Configuration job) 
  103.     throws IOException{
  104.     this.maxLineLength = job.getInt("mapred.linerecordreader.maxlength",
  105.                                     Integer.MAX_VALUE);
  106.     this.in = new LineReader(in, job);
  107.     this.start = offset;
  108.     this.pos = offset;
  109.     this.end = endOffset;    
  110.   }
  111.   
  112.   public LongWritable createKey() {
  113.     return new LongWritable();
  114.   }
  115.   
  116.   public Text createValue() {
  117.     return new Text();
  118.   }
  119.   
  120.   /** Read a line. */
  121.   public synchronized boolean next(LongWritable key, Text value)
  122.     throws IOException {
  123.     while (pos < end) {
  124.       key.set(pos);
  125.       int newSize = in.readLine(value, maxLineLength,
  126.                                 Math.max((int)Math.min(Integer.MAX_VALUE, end-pos),
  127.                                          maxLineLength));
  128.       if (newSize == 0) {
  129.         return false;
  130.       }
  131.       pos += newSize;
  132.       if (newSize < maxLineLength) {
  133.         return true;
  134.       }
  135.       // line too long. try again
  136.       LOG.info("Skipped line of size " + newSize + " at pos " + (pos - newSize));
  137.     }
  138.     return false;
  139.   }
  140.   /**
  141.    * Get the progress within the split
  142.    */
  143.   public float getProgress() {
  144.     if (start == end) {
  145.       return 0.0f;
  146.     } else {
  147.       return Math.min(1.0f, (pos - start) / (float)(end - start));
  148.     }
  149.   }
  150.   
  151.   public  synchronized long getPos() throws IOException {
  152.     return pos;
  153.   }
  154.   public synchronized void close() throws IOException {
  155.     if (in != null) {
  156.       in.close(); 
  157.     }
  158.   }
  159. }