title: Hadoop Map Reduce CustomSplit/CustomRecordReader
body: <p>I have a huge text file and I wanted to split the file so that each chunk has 5 lines. I implemented my own GWASInputFormat and GWASRecordReader classes. However my question is, in the following code(which I copied from <a href="http://bigdatacircus.com/2012/08/01/wordcount-with-custom-record-reader-of-textinputformat/" rel="nofollow">http://bigdatacircus.com/2012/08/01/wordcount-with-custom-record-reader-of-textinputformat/</a>), inside the initialize() method I have the following lines</p>
  
  <pre><code>FileSplit split = (FileSplit) genericSplit;
  final Path file = split.getPath();
  Configuration conf = context.getConfiguration();
  </code></pre>
  
  <p>My question is,  Is the file already split by the time the initialize() method is called in my GWASRecordReader class? I thought that I was doing it(the split) in the GWASRecordReader class. Let me know if my thought process is right here.</p>
  
  <pre><code>package com.test;
  
  import java.io.IOException;
  
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.FSDataInputStream;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.LongWritable;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapreduce.InputSplit;
  import org.apache.hadoop.mapreduce.RecordReader;
  import org.apache.hadoop.mapreduce.TaskAttemptContext;
  import org.apache.hadoop.mapreduce.lib.input.FileSplit;
  import org.apache.hadoop.util.LineReader;
  
  public class GWASRecordReader extends RecordReader&lt;LongWritable, Text&gt; {
  
  private final int NLINESTOPROCESS = 5;
  private LineReader in;
  private LongWritable key;
  private Text value = new Text();
  private long start = 0;
  private long pos = 0;
  private long end = 0;
  private int maxLineLength;
  
  public void close() throws IOException {
      if(in != null) {
          in.close();
      }
  }
  
  public LongWritable getCurrentKey() throws IOException, InterruptedException {
      return key;
  }
  
  public Text getCurrentValue() throws IOException, InterruptedException {
      return value;
  }
  
  public float getProgress() throws IOException, InterruptedException {
      if(start == end) {
          return 0.0f;
      }
      else {
          return Math.min(1.0f, (pos - start)/(float) (end - start));
      }
  }
  
  public void initialize(InputSplit genericSplit, TaskAttemptContext context) throws IOException {
      FileSplit split = (FileSplit) genericSplit;
      final Path file = split.getPath();
      Configuration conf = context.getConfiguration();
      this.maxLineLength = conf.getInt("mapred.linerecordreader.maxlength",Integer.MAX_VALUE);
      FileSystem fs = file.getFileSystem(conf);
      start = split.getStart();
      end = start + split.getLength();
      System.out.println("---------------SPLIT LENGTH---------------------" + split.getLength());
      boolean skipFirstLine = false;
      FSDataInputStream filein = fs.open(split.getPath());
  
      if(start != 0) {
          skipFirstLine = true;
          --start;
          filein.seek(start);
      }
  
      in = new LineReader(filein, conf);
      if(skipFirstLine) {
          start += in.readLine(new Text(),0,(int)Math.min((long)Integer.MAX_VALUE, end - start));
      }
      this.pos = start;
  }
  
  public boolean nextKeyValue() throws IOException, InterruptedException {
      if (key == null) {
          key = new LongWritable();
      }
  
      key.set(pos);
  
      if (value == null) {
          value = new Text();
      }
      value.clear();
      final Text endline = new Text("\n");
      int newSize = 0;
      for(int i=0; i&lt;NLINESTOPROCESS;i++) {
          Text v = new Text();
          while( pos &lt; end) {
              newSize = in.readLine(v ,maxLineLength, Math.max((int)Math.min(Integer.MAX_VALUE, end - pos), maxLineLength));
              value.append(v.getBytes(), 0, v.getLength());
              value.append(endline.getBytes(),0,endline.getLength());
              if(newSize == 0) {
                  break;
              }
              pos += newSize;
              if(newSize &lt; maxLineLength) {
                  break;
              }
          }
      }
  
      if(newSize == 0) {
          key = null;
          value = null;
          return false;
      } else {
          return true;
      }
  }
  }
  </code></pre>