/** Constructs a split with host and cached-blocks information
 *
 * @param file the file name
 * @param start the position of the first byte in the file to process
 * @param length the number of bytes in the file to process
 * @param hosts the list of hosts containing the block
 * @param inMemoryHosts the list of hosts containing the block in memory
 */
public FileSplit(Path file, long start, long length, String[] hosts,
    String[] inMemoryHosts) {
  this(file, start, length, hosts);
  hostInfos = new SplitLocationInfo[hosts.length];
  for (int i = 0; i < hosts.length; i++) {
    // because N will be tiny, scanning is probably faster than a HashSet
    boolean inMemory = false;
    for (String inMemoryHost : inMemoryHosts) {
      if (inMemoryHost.equals(hosts[i])) {
        inMemory = true;
        break;
      }
    }
    hostInfos[i] = new SplitLocationInfo(hosts[i], inMemory);
  }
}
 

@Test
public void testSplitLocationInfo() throws Exception {
  Configuration conf = getConfiguration();
  conf.set(org.apache.hadoop.mapreduce.lib.input.FileInputFormat.INPUT_DIR,
      "test:///a1/a2");
  Job job = Job.getInstance(conf);
  TextInputFormat fileInputFormat = new TextInputFormat();
  List<InputSplit> splits = fileInputFormat.getSplits(job);
  String[] locations = splits.get(0).getLocations();
  Assert.assertEquals(2, locations.length);
  SplitLocationInfo[] locationInfo = splits.get(0).getLocationInfo();
  Assert.assertEquals(2, locationInfo.length);
  SplitLocationInfo localhostInfo = locations[0].equals("localhost") ?
      locationInfo[0] : locationInfo[1];
  SplitLocationInfo otherhostInfo = locations[0].equals("otherhost") ?
      locationInfo[0] : locationInfo[1];
  Assert.assertTrue(localhostInfo.isOnDisk());
  Assert.assertTrue(localhostInfo.isInMemory());
  Assert.assertTrue(otherhostInfo.isOnDisk());
  Assert.assertFalse(otherhostInfo.isInMemory());
}
 

/** Constructs a split with host and cached-blocks information
 *
 * @param file the file name
 * @param start the position of the first byte in the file to process
 * @param length the number of bytes in the file to process
 * @param hosts the list of hosts containing the block
 * @param inMemoryHosts the list of hosts containing the block in memory
 */
public FileSplit(Path file, long start, long length, String[] hosts,
    String[] inMemoryHosts) {
  this(file, start, length, hosts);
  hostInfos = new SplitLocationInfo[hosts.length];
  for (int i = 0; i < hosts.length; i++) {
    // because N will be tiny, scanning is probably faster than a HashSet
    boolean inMemory = false;
    for (String inMemoryHost : inMemoryHosts) {
      if (inMemoryHost.equals(hosts[i])) {
        inMemory = true;
        break;
      }
    }
    hostInfos[i] = new SplitLocationInfo(hosts[i], inMemory);
  }
}
 

@Test
public void testSplitLocationInfo() throws Exception {
  Configuration conf = getConfiguration();
  conf.set(org.apache.hadoop.mapreduce.lib.input.FileInputFormat.INPUT_DIR,
      "test:///a1/a2");
  Job job = Job.getInstance(conf);
  TextInputFormat fileInputFormat = new TextInputFormat();
  List<InputSplit> splits = fileInputFormat.getSplits(job);
  String[] locations = splits.get(0).getLocations();
  Assert.assertEquals(2, locations.length);
  SplitLocationInfo[] locationInfo = splits.get(0).getLocationInfo();
  Assert.assertEquals(2, locationInfo.length);
  SplitLocationInfo localhostInfo = locations[0].equals("localhost") ?
      locationInfo[0] : locationInfo[1];
  SplitLocationInfo otherhostInfo = locations[0].equals("otherhost") ?
      locationInfo[0] : locationInfo[1];
  Assert.assertTrue(localhostInfo.isOnDisk());
  Assert.assertTrue(localhostInfo.isInMemory());
  Assert.assertTrue(otherhostInfo.isOnDisk());
  Assert.assertFalse(otherhostInfo.isInMemory());
}
 

public HashMap<String, ArrayList<Integer>> getLocationsOfSplits() throws IOException {
    HashMap<String, ArrayList<Integer>> splitsMap = new HashMap<>();
    ArrayList<Integer> temp;
    int i = 0;
    String hostname;
    for (InputSplit s : this.splits) {
        SplitLocationInfo[] info = s.getLocationInfo();
        hostname = info[0].getLocation();
        if (splitsMap.containsKey(hostname)) {
            temp = splitsMap.get(hostname);
            temp.add(i);
        } else {
            temp = new ArrayList<>();
            temp.add(i);
            splitsMap.put(hostname, temp);
        }
        i++;
    }

    return splitsMap;
}
 

@Override
@Evolving
public SplitLocationInfo[] getLocationInfo() throws IOException {
  return hostInfos;
}
 

@Override
@Evolving
public SplitLocationInfo[] getLocationInfo() throws IOException {
  return hostInfos;
}
 

/**
 * Gets info about which nodes the input split is stored on and how it is
 * stored at each location.
 * 
 * @return list of <code>SplitLocationInfo</code>s describing how the split
 *    data is stored at each location. A null value indicates that all the
 *    locations have the data stored on disk.
 * @throws IOException
 */
@Evolving
public SplitLocationInfo[] getLocationInfo() throws IOException {
  return null;
}
 

/**
 * Gets info about which nodes the input split is stored on and how it is
 * stored at each location.
 * 
 * @return list of <code>SplitLocationInfo</code>s describing how the split
 *    data is stored at each location. A null value indicates that all the
 *    locations have the data stored on disk.
 * @throws IOException
 */
@Evolving
public SplitLocationInfo[] getLocationInfo() throws IOException {
  return null;
}