multi-input-crunch-source-impl

MultiSequenceFileInputFormat.java

public class MultiSequenceFileInputFormat<K, V> extends CombineFileInputFormat<K, V> {

  @SuppressWarnings("unchecked")
  @Override
  public RecordReader<K, V> createRecordReader(InputSplit split,
    TaskAttemptContext context) throws IOException {
    return new CombineFileRecordReader(
      (CombineFileSplit) split, context, MultiSequenceFileRecordReader.class
    );
  }

}

MultiSequenceFileRecordReader.java

public final class MultiSequenceFileRecordReader<K, V> extends RecordReader<K, V> {

  private CombineFileSplit split;
  private TaskAttemptContext context;
  private int index;
  private RecordReader<K, V> delegate;

  public MultiSequenceFileRecordReader(CombineFileSplit split, TaskAttemptContext context,
    Integer index) throws IOException {
    this.split = split;
    this.context = context;
    this.index = index;
    delegate = new SequenceFileRecordReader<K, V>();
  }


  public void initialize(InputSplit genericSplit, TaskAttemptContext context)
    throws IOException, InterruptedException {
    this.split = (CombineFileSplit) genericSplit;
    this.context = context;

    if (null == delegate) {
      delegate = new SequenceFileRecordReader<K, V>();
    }

    FileSplit fileSplit = new FileSplit(
      this.split.getPath(index),
      this.split.getOffset(index),
      this.split.getLength(index),
      this.split.getLocations()
    );

    this.delegate.initialize(fileSplit, this.context);
  }

  public boolean nextKeyValue() throws IOException, InterruptedException {
    return delegate.nextKeyValue();
  }

  @Override
  public K getCurrentKey() throws IOException, InterruptedException {
    return delegate.getCurrentKey();
  }

  @Override
  public V getCurrentValue() throws IOException, InterruptedException {
    return delegate.getCurrentValue();
  }

  @Override
  public float getProgress() throws IOException, InterruptedException {
    return delegate.getProgress();
  }

  public synchronized void close() throws IOException {
    if (delegate != null) {
      delegate.close();
      delegate = null;
    }
  }

}

TrainingRecordSource.java

public class TrainingRecordSource implements TableSource<ShortWritable, BytesWritable> {

  @Override
  public PTableType<ShortWritable, BytesWritable> getTableType() {
    return Writables.tableOf(
      Writables.writables(ShortWritable.class), Writables.writables(BytesWritable.class)
    );
  }

  @Override
  public PType<Pair<ShortWritable, BytesWritable>> getType() {
    return getTableType();
  }

  @Override
  public void configureSource(Job job, int i) throws IOException {
    job.setInputFormatClass(MultiSequenceFileInputFormat.class);
    //traverse fs and add paths as inputs ...
    for (FileStatus input : inputs) {
      MultiSequenceFileInputFormat.addInputPath(job, input.getPath());
    }
  }

  @Override
  public long getSize(Configuration configuration) {
    return Integer.MAX_VALUE;
  }

}