@Override
  public List<InputSplit> getSplits(JobContext context) throws IOException {
    final Configuration conf = context.getConfiguration();
    final KijiURI inputTableURI = getInputTableURI(conf);
    final Kiji kiji = Kiji.Factory.open(inputTableURI, conf);
    final KijiTable table = kiji.openTable(inputTableURI.getTable());
    final HTableInterface htable = HBaseKijiTable.downcast(table).openHTableConnection();
    try {
      final List<InputSplit> splits = Lists.newArrayList();
      for (KijiRegion region : table.getRegions()) {
        final byte[] startKey = region.getStartKey();
        // TODO: a smart way to get which location is most relevant.
        final String location =
            region.getLocations().isEmpty() ? null : region.getLocations().iterator().next();
        final TableSplit tableSplit = new TableSplit(
            htable.getTableName(), startKey, region.getEndKey(), location);
        splits.add(new KijiTableSplit(tableSplit, startKey));
      }
      return splits;
    } finally {
      htable.close();
    }
  }
@Override
  public List<InputSplit> getSplits(JobContext context) throws IOException {
    final Configuration conf = context.getConfiguration();
    final KijiURI inputTableURI = getInputTableURI(conf);
    final Kiji kiji = Kiji.Factory.open(inputTableURI, conf);
    final KijiTable table = kiji.openTable(inputTableURI.getTable());
    final HTableInterface htable = HBaseKijiTable.downcast(table).openHTableConnection();
    try {
      final List<InputSplit> splits = Lists.newArrayList();
      for (KijiRegion region : table.getRegions()) {
        final byte[] startKey = region.getStartKey();
       
        final String location =
            region.getLocations().isEmpty() ? null : region.getLocations().iterator().next();
        final TableSplit tableSplit = new TableSplit(
            htable.getTableName(), startKey, region.getEndKey(), location);
        splits.add(new KijiTableSplit(tableSplit, startKey));
      }
      return splits;
    } finally {
      htable.close();
    }
  }
@override public list<inputsplit> getsplits(jobcontext context) throws ioexception { final configuration conf = context.getconfiguration(); final kijiuri inputtableuri = getinputtableuri(conf); final kiji kiji = kiji.factory.open(inputtableuri, conf); final kijitable table = kiji.opentable(inputtableuri.gettable()); final htableinterface htable = hbasekijitable.downcast(table).openhtableconnection(); try { final list<inputsplit> splits = lists.newarraylist(); for (kijiregion region : table.getregions()) { final byte[] startkey = region.getstartkey(); final string location = region.getlocations().isempty() ? null : region.getlocations().iterator().next(); final tablesplit tablesplit = new tablesplit( htable.gettablename(), startkey, region.getendkey(), location); splits.add(new kijitablesplit(tablesplit, startkey)); } return splits; } finally { htable.close(); } }