code: public InputSplit[] getSplits(JobConf job, int numSplits)
    throws IOException {
      ArrayList<FileSplit> splits = new ArrayList<FileSplit>();
      for (FileStatus status : listLocatedStatus(job)) {
        Path fileName = status.getPath();
        if (status.isDir()) {
          throw new IOException("Not a file: " + fileName);
        }
        FileSystem  fs = fileName.getFileSystem(job);
        LineReader lr = null;
        try {
          FSDataInputStream in  = fs.open(fileName);
          lr = new LineReader(in, job);
          Text line = new Text();
          int numLines = 0;
          long begin = 0;
          long length = 0;
          int num = -1;
          while ((num = lr.readLine(line)) > 0) {
            numLines++;
            length += num;
            if (numLines == N) {
              splits.add(new FileSplit(fileName, begin, length, new String[]{}));
              begin += length;
              length = 0;
              numLines = 0;
            }
          }
          if (numLines != 0) {
            splits.add(new FileSplit(fileName, begin, length, new String[]{}));
          }
     
        } finally {
          if (lr != null) {
            lr.close();
          }
        }
      }
      return splits.toArray(new FileSplit[splits.size()]);
    }
docs: Logically splits the set of input files for the job, splits N lines
  of the input as one split.
  
  @see org.apache.hadoop.mapred.FileInputFormat#getSplits(JobConf, int)