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 */
018package org.apache.hadoop.mapreduce.lib.input;
019
020import java.io.IOException;
021
022import org.apache.hadoop.classification.InterfaceAudience;
023import org.apache.hadoop.classification.InterfaceStability;
024import org.apache.hadoop.conf.Configuration;
025import org.apache.hadoop.fs.FileSystem;
026import org.apache.hadoop.fs.Path;
027import org.apache.hadoop.io.BytesWritable;
028import org.apache.hadoop.io.DataOutputBuffer;
029import org.apache.hadoop.io.SequenceFile;
030import org.apache.hadoop.mapreduce.InputSplit;
031import org.apache.hadoop.mapreduce.RecordReader;
032import org.apache.hadoop.mapreduce.TaskAttemptContext;
033
034/**
035 * InputFormat reading keys, values from SequenceFiles in binary (raw)
036 * format.
037 */
038@InterfaceAudience.Public
039@InterfaceStability.Stable
040public class SequenceFileAsBinaryInputFormat
041    extends SequenceFileInputFormat<BytesWritable,BytesWritable> {
042
043  public SequenceFileAsBinaryInputFormat() {
044    super();
045  }
046
047  public RecordReader<BytesWritable,BytesWritable> createRecordReader(
048      InputSplit split, TaskAttemptContext context)
049      throws IOException {
050    return new SequenceFileAsBinaryRecordReader();
051  }
052
053  /**
054   * Read records from a SequenceFile as binary (raw) bytes.
055   */
056  public static class SequenceFileAsBinaryRecordReader
057      extends RecordReader<BytesWritable,BytesWritable> {
058    private SequenceFile.Reader in;
059    private long start;
060    private long end;
061    private boolean done = false;
062    private DataOutputBuffer buffer = new DataOutputBuffer();
063    private SequenceFile.ValueBytes vbytes;
064    private BytesWritable key = null;
065    private BytesWritable value = null;
066
067    public void initialize(InputSplit split, TaskAttemptContext context) 
068        throws IOException, InterruptedException {
069      Path path = ((FileSplit)split).getPath();
070      Configuration conf = context.getConfiguration();
071      FileSystem fs = path.getFileSystem(conf);
072      this.in = new SequenceFile.Reader(fs, path, conf);
073      this.end = ((FileSplit)split).getStart() + split.getLength();
074      if (((FileSplit)split).getStart() > in.getPosition()) {
075        in.sync(((FileSplit)split).getStart());    // sync to start
076      }
077      this.start = in.getPosition();
078      vbytes = in.createValueBytes();
079      done = start >= end;
080    }
081    
082    @Override
083    public BytesWritable getCurrentKey() 
084        throws IOException, InterruptedException {
085      return key;
086    }
087    
088    @Override
089    public BytesWritable getCurrentValue() 
090        throws IOException, InterruptedException {
091      return value;
092    }
093
094    /**
095     * Retrieve the name of the key class for this SequenceFile.
096     * @see org.apache.hadoop.io.SequenceFile.Reader#getKeyClassName
097     */
098    public String getKeyClassName() {
099      return in.getKeyClassName();
100    }
101
102    /**
103     * Retrieve the name of the value class for this SequenceFile.
104     * @see org.apache.hadoop.io.SequenceFile.Reader#getValueClassName
105     */
106    public String getValueClassName() {
107      return in.getValueClassName();
108    }
109
110    /**
111     * Read raw bytes from a SequenceFile.
112     */
113    public synchronized boolean nextKeyValue()
114        throws IOException, InterruptedException {
115      if (done) {
116        return false;
117      }
118      long pos = in.getPosition();
119      boolean eof = -1 == in.nextRawKey(buffer);
120      if (!eof) {
121        if (key == null) {
122          key = new BytesWritable();
123        }
124        if (value == null) {
125          value = new BytesWritable();
126        }
127        key.set(buffer.getData(), 0, buffer.getLength());
128        buffer.reset();
129        in.nextRawValue(vbytes);
130        vbytes.writeUncompressedBytes(buffer);
131        value.set(buffer.getData(), 0, buffer.getLength());
132        buffer.reset();
133      }
134      return !(done = (eof || (pos >= end && in.syncSeen())));
135    }
136
137    public void close() throws IOException {
138      in.close();
139    }
140
141    /**
142     * Return the progress within the input split
143     * @return 0.0 to 1.0 of the input byte range
144     */
145    public float getProgress() throws IOException, InterruptedException {
146      if (end == start) {
147        return 0.0f;
148      } else {
149        return Math.min(1.0f, (float)((in.getPosition() - start) /
150                                      (double)(end - start)));
151      }
152    }
153  }
154}