PigStorage

gistfile1.java

import com.google.common.base.Preconditions;
import com.ioffer.hadoop.input.LimitedDailyLogFormat;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigTextInputFormat;
import org.apache.pig.builtin.PigStorage;

import java.io.IOException;

public class LimitedDailyLogLoader extends PigStorage {

  private Integer limit;
  private static final String DELIMITER = Character.toString('\t');

  public LimitedDailyLogLoader(String limit) {
    this(Integer.parseInt(limit), DELIMITER);
  }

  public LimitedDailyLogLoader(String limit, String delimeter) {
    this(Integer.parseInt(limit), delimeter);
  }

  public LimitedDailyLogLoader(Integer limit, String delimiter) {
    super(delimiter);
    Preconditions.checkArgument(limit != null && limit > 0, "limit is null or less than zero");
    this.limit = limit;
  }

  @Override
  public InputFormat getInputFormat() {
    return new PigTextInputFormat();
  }

  @Override
  public void setLocation(String location, Job job) throws IOException {
    LimitedDailyLogFormat.setInputPaths(job, location, limit);
  }

}