public void setupRunner(String jobName, Class<?> runnerClass,
    Class<? extends TableMapper<?, ?>> mapperClass, Class<? extends Reducer<?, ?, ?, ?>> reducerClass,
    Class<? extends WritableComparable<?>> mapOutputKeyClass,
    Class<? extends Writable> mapOutputValueClass,
    Class<? extends WritableComparable<?>> outputKeyClass,
    Class<? extends Writable> outputValueClass,
    Class<? extends OutputFormat<?, ?>> outputFormatClass) {

  this.jobName = jobName;
  this.runnerClass = runnerClass;
  this.mapperClass = mapperClass;
  this.reducerClass = reducerClass;
  this.mapOutputKeyClass = mapOutputKeyClass;
  this.mapOutputValueClass = mapOutputValueClass;
  this.outputKeyClass = outputKeyClass;
  this.outputValueClass = outputValueClass;
  this.outputFormatClass = outputFormatClass;
  this.isCallSetUpRunnerMethod = true;
}
 

public void setupRunner(String jobName, Class<?> runnerClass,
    Class<? extends TableMapper<?, ?>> mapperClass, Class<? extends Reducer<?, ?, ?, ?>> reducerClass,
    Class<? extends WritableComparable<?>> outputKeyClass,
    Class<? extends Writable> outputValueClass,
    Class<? extends OutputFormat<?, ?>> outputFormatClass) {
    this.setupRunner(jobName, runnerClass, mapperClass, reducerClass, outputKeyClass, outputValueClass, outputKeyClass, outputValueClass, outputFormatClass);

}
 

public void setupRunner(String jobName, Class<?> runnerClass,
    Class<? extends TableMapper<?, ?>> mapperClass, Class<? extends Reducer<?, ?, ?, ?>> reducerClass,
    Class<? extends WritableComparable<?>> mapOutputKeyClass,
    Class<? extends Writable> mapOutputValueClass,
    Class<? extends WritableComparable<?>> outputKeyClass,
    Class<? extends Writable> outputValueClass) {
  this.setupRunner(jobName, runnerClass, mapperClass, reducerClass, mapOutputKeyClass, mapOutputValueClass, outputKeyClass, outputValueClass, TransformerOutputFormat.class);
}
 

/**
 * 具体设置参数
 * 
 * @param jobName
 * @param runnerClass
 * @param mapperClass
 * @param reducerClass
 * @param mapOutputKeyClass
 * @param mapOutputValueClass
 * @param outputKeyClass
 * @param outputValueClass
 * @param outputFormatClass
 */
public void setupRunner(String jobName, Class<?> runnerClass, Class<? extends TableMapper<?, ?>> mapperClass, Class<? extends Reducer<?, ?, ?, ?>> reducerClass, Class<? extends WritableComparable<?>> mapOutputKeyClass, Class<? extends Writable> mapOutputValueClass, Class<? extends WritableComparable<?>> outputKeyClass, Class<? extends Writable> outputValueClass,
        Class<? extends OutputFormat<?, ?>> outputFormatClass) {
    this.jobName = jobName;
    this.runnerClass = runnerClass;
    this.mapperClass = mapperClass;
    this.reducerClass = reducerClass;
    this.mapOutputKeyClass = mapOutputKeyClass;
    this.mapOutputValueClass = mapOutputValueClass;
    this.outputKeyClass = outputKeyClass;
    this.outputValueClass = outputValueClass;
    this.outputFormatClass = outputFormatClass;
    this.isCallSetUpRunnerMethod = true;
}
 

@Override
public Class[] getHbaseDependencyClasses() {
  return new Class[] {
    HConstants.class, org.apache.hadoop.hbase.protobuf.generated.ClientProtos.class,
    org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.class, Put.class,
    RpcServer.class, CompatibilityFactory.class, JobUtil.class, TableMapper.class, FastLongHistogram.class,
    Snapshot.class, ZooKeeper.class, Channel.class, Message.class, UnsafeByteOperations.class, Lists.class,
    Tracer.class, MetricRegistry.class, ArrayUtils.class, ObjectMapper.class, Versioned.class,
    JsonView.class, ZKWatcher.class, CacheLoader.class
  };
}
 

@Override
public Class[] getHbaseDependencyClasses() {
  return new Class[] {
    HConstants.class, ClientProtos.class, Put.class, CompatibilityFactory.class, TableMapper.class,
    ZooKeeper.class, Channel.class, Message.class, Lists.class, Trace.class, MetricsRegistry.class
  };
}
 

public Class[] getHbaseDependencyClasses() {
  return new Class[] {
    HConstants.class, ClientProtos.class, Put.class, CompatibilityFactory.class,
    JobUtil.class, TableMapper.class, FastLongHistogram.class, Snapshot.class,
    ZooKeeper.class, Channel.class, Message.class, Lists.class, Trace.class, MetricsRegistry.class
  };
}
 

@Override
public Class[] getHbaseDependencyClasses() {
  return new Class[] {
    HConstants.class, ClientProtos.class, Put.class, CompatibilityFactory.class, TableMapper.class,
    ZooKeeper.class, Channel.class, Message.class, Lists.class, Trace.class, MetricsRegistry.class
  };
}
 

@Override
public Class[] getHbaseDependencyClasses() {
  return new Class[] {
    HConstants.class, org.apache.hadoop.hbase.protobuf.generated.ClientProtos.class,
    org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.class, Put.class,
    RpcServer.class, CompatibilityFactory.class, JobUtil.class, TableMapper.class, FastLongHistogram.class,
    Snapshot.class, ZooKeeper.class, Channel.class, Message.class, UnsafeByteOperations.class, Lists.class,
    Tracer.class, MetricRegistry.class, ArrayUtils.class, ObjectMapper.class, Versioned.class, JsonView.class,
    ZKWatcher.class
  };
}
 

@Override
public Class[] getHbaseDependencyClasses() {
  return new Class[] {
    HConstants.class, org.apache.hadoop.hbase.protobuf.generated.ClientProtos.class,
    org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.class, Put.class,
    RpcServer.class, CompatibilityFactory.class, JobUtil.class, TableMapper.class, FastLongHistogram.class,
    Snapshot.class, ZooKeeper.class, Channel.class, Message.class, UnsafeByteOperations.class, Lists.class,
    Tracer.class, MetricRegistry.class, ArrayUtils.class, ObjectMapper.class, Versioned.class, JsonView.class,
    ZKWatcher.class
  };
}
 

@Override
public Class[] getHbaseDependencyClasses() {
  return new Class[]{
    HConstants.class, ClientProtos.class, Put.class, CompatibilityFactory.class, TableMapper.class,
    ZooKeeper.class, Channel.class, Message.class, Lists.class, Trace.class, MetricsRegistry.class
  };
}
 

private TableInputFormat getDelegate(Configuration conf) throws IOException {
  TableInputFormat delegate = new TableInputFormat();
  String tableName = HBaseMetadataProvider.getTableName(dataset.getName());
  conf.set(TableInputFormat.INPUT_TABLE, tableName);
  if (view != null) {
    Job tempJob = new Job();
    Scan scan = ((BaseEntityScanner) view.newEntityScanner()).getScan();
    TableMapReduceUtil.initTableMapperJob(tableName, scan, TableMapper.class, null,
        null, tempJob);
    Configuration tempConf = Hadoop.JobContext.getConfiguration.invoke(tempJob);
    conf.set(SCAN, tempConf.get(SCAN));
  }
  delegate.setConf(conf);
  return delegate;
}
 

@SuppressWarnings("rawtypes")
public static Class<? extends TableMapper> wrap(String type) {
	Class<? extends TableMapper> c = IndexMapper.class;
	if (null != type && type.length() > 0) {
		if (type.equals(Const.MAPPER_TYPE_JSON))
			c = IndexJsonMapper.class;
		else if (type.equals(Const.MAPPER_TYPE_ROWKEY))
			c = IndexRowkeyMapper.class;
	}
	return c;
}
 

public void setupRunner(String jobName, Class<?> runnerClass,
    Class<? extends TableMapper<?, ?>> mapperClass, Class<? extends Reducer<?, ?, ?, ?>> reducerClass,
    Class<? extends WritableComparable<?>> outputKeyClass,
    Class<? extends Writable> outputValueClass) {
  this.setupRunner(jobName, runnerClass, mapperClass, reducerClass, outputKeyClass, outputValueClass, TransformerOutputFormat.class);
}
 

/**
 * 设置job参数
 * 
 * @param jobName
 *            job名称
 * @param runnerClass
 *            runne class
 * @param mapperClass
 *            mapper的class
 * @param reducerClass
 *            reducer的class
 * @param outputKeyClass
 *            输出key
 * @param outputValueClass
 *            输出value
 */
public void setupRunner(String jobName, Class<?> runnerClass, Class<? extends TableMapper<?, ?>> mapperClass, Class<? extends Reducer<?, ?, ?, ?>> reducerClass, Class<? extends WritableComparable<?>> outputKeyClass, Class<? extends Writable> outputValueClass) {
    this.setupRunner(jobName, runnerClass, mapperClass, reducerClass, outputKeyClass, outputValueClass, outputKeyClass, outputValueClass, TransformerOutputFormat.class);
}
 

/**
 * 设置参数
 * 
 * @param jobName
 * @param runnerClass
 * @param mapperClass
 * @param reducerClass
 * @param outputKeyClass
 * @param outputValueClass
 * @param outputFormatClass
 */
public void setupRunner(String jobName, Class<?> runnerClass, Class<? extends TableMapper<?, ?>> mapperClass, Class<? extends Reducer<?, ?, ?, ?>> reducerClass, Class<? extends WritableComparable<?>> outputKeyClass, Class<? extends Writable> outputValueClass, Class<? extends OutputFormat<?, ?>> outputFormatClass) {
    this.setupRunner(jobName, runnerClass, mapperClass, reducerClass, outputKeyClass, outputValueClass, outputKeyClass, outputValueClass, outputFormatClass);
}
 

/**
 * 设置参数
 * 
 * @param jobName
 * @param runnerClass
 * @param mapperClass
 * @param reducerClass
 * @param mapOutputKeyClass
 * @param mapOutputValueClass
 * @param outputKeyClass
 * @param outputValueClass
 */
public void setupRunner(String jobName, Class<?> runnerClass, Class<? extends TableMapper<?, ?>> mapperClass, Class<? extends Reducer<?, ?, ?, ?>> reducerClass, Class<? extends WritableComparable<?>> mapOutputKeyClass, Class<? extends Writable> mapOutputValueClass, Class<? extends WritableComparable<?>> outputKeyClass, Class<? extends Writable> outputValueClass) {
    this.setupRunner(jobName, runnerClass, mapperClass, reducerClass, mapOutputKeyClass, mapOutputValueClass, outputKeyClass, outputValueClass, TransformerOutputFormat.class);
}