private Vector<FilenameInputSplit> getPathInputSplits(FileSystem fs, Vector<FileStatus> files, long chunkSize)
      throws IOException {
        Vector<FilenameInputSplit> splits = new Vector<FilenameInputSplit>();
        splits.add(new FilenameInputSplit());
        for (FileStatus status : files) {
            if (status.isDir()) {
                System.out.println("Skipping directory: " + status.getPath().toString());
                continue;
            }

            System.out.println(String.format("Adding file %s of length %d to split %d",
                                             status.getPath().toString(),
                                             status.getLen(),
                                             splits.size()));

            Set<String> hosts = new TreeSet<String>();

            try {
                for (BlockLocation location : fs.getFileBlockLocations(status, 0, status.getLen()))
                    Collections.addAll(hosts, location.getHosts());

            } catch (IOException e) {
                //noinspection ThrowableInstanceNeverThrown
                new RuntimeException(e);
            }

            splits.lastElement().insertPath(status.getPath(), hosts, status.getLen());

            if (splits.lastElement().getLength() > chunkSize)
                splits.add(new FilenameInputSplit());
        }

        Collections.sort(files, new Comparator<FileStatus>() {
            @Override
            public int compare(FileStatus fileStatus, FileStatus fileStatus2) {
                Long size = fileStatus.getLen();
                return size.compareTo(fileStatus2.getLen());
            }
        });

        while (splits.size() > 0 && splits.lastElement().getLength() == 0)
            splits.remove(splits.lastElement());

        return splits;
    }