@Test public void testBalanceSplit() throws Exception { List<Pair<SnapshotFileInfo, Long>> files = new ArrayList<Pair<SnapshotFileInfo, Long>>(); for (long i = 0; i <= 20; i++) { SnapshotFileInfo fileInfo = SnapshotFileInfo.newBuilder() .setType(SnapshotFileInfo.Type.HFILE) .setHfile("file-" + i) .build(); files.add(new Pair<SnapshotFileInfo, Long>(fileInfo, i)); } List<List<Pair<SnapshotFileInfo, Long>>> splits = ExportSnapshot.getBalancedSplits(files, 5); assertEquals(5, splits.size()); String[] split0 = new String[] {"file-20", "file-11", "file-10", "file-1", "file-0"}; verifyBalanceSplit(splits.get(0), split0, 42); String[] split1 = new String[] {"file-19", "file-12", "file-9", "file-2"}; verifyBalanceSplit(splits.get(1), split1, 42); String[] split2 = new String[] {"file-18", "file-13", "file-8", "file-3"}; verifyBalanceSplit(splits.get(2), split2, 42); String[] split3 = new String[] {"file-17", "file-14", "file-7", "file-4"}; verifyBalanceSplit(splits.get(3), split3, 42); String[] split4 = new String[] {"file-16", "file-15", "file-6", "file-5"}; verifyBalanceSplit(splits.get(4), split4, 42); }
static List<List<Pair<SnapshotFileInfo, Long>>> getBalancedSplits( final List<Pair<SnapshotFileInfo, Long>> files, final int ngroups) { Collections.sort(files, new Comparator<Pair<SnapshotFileInfo, Long>>() { public int compare(Pair<SnapshotFileInfo, Long> a, Pair<SnapshotFileInfo, Long> b) { long r = a.getSecond() - b.getSecond(); return (r < 0) ? -1 : ((r > 0) ? 1 : 0); } }); List<List<Pair<SnapshotFileInfo, Long>>> fileGroups = new LinkedList<List<Pair<SnapshotFileInfo, Long>>>(); long[] sizeGroups = new long[ngroups]; int hi = files.size() - 1; int lo = 0; List<Pair<SnapshotFileInfo, Long>> group; int dir = 1; int g = 0; while (hi >= lo) { if (g == fileGroups.size()) { group = new LinkedList<Pair<SnapshotFileInfo, Long>>(); fileGroups.add(group); } else { group = fileGroups.get(g); } Pair<SnapshotFileInfo, Long> fileInfo = files.get(hi--); sizeGroups[g] += fileInfo.getSecond(); group.add(fileInfo); g += dir; if (g == ngroups) { dir = -1; g = ngroups - 1; } else if (g < 0) { dir = 1; g = 0; } } if (LOG.isDebugEnabled()) { for (int i = 0; i < sizeGroups.length; ++i) { LOG.debug("export split=" + i + " size=" + StringUtils.humanReadableInt(sizeGroups[i])); } } return fileGroups; }
ExportSnapshot extends Configured implements Tool { static List<List<Pair<SnapshotFileInfo, Long>>> getBalancedSplits( final List<Pair<SnapshotFileInfo, Long>> files, final int ngroups) { Collections.sort(files, new Comparator<Pair<SnapshotFileInfo, Long>>() { public int compare(Pair<SnapshotFileInfo, Long> a, Pair<SnapshotFileInfo, Long> b) { long r = a.getSecond() - b.getSecond(); return (r < 0) ? -1 : ((r > 0) ? 1 : 0); } }); List<List<Pair<SnapshotFileInfo, Long>>> fileGroups = new LinkedList<List<Pair<SnapshotFileInfo, Long>>>(); long[] sizeGroups = new long[ngroups]; int hi = files.size() - 1; int lo = 0; List<Pair<SnapshotFileInfo, Long>> group; int dir = 1; int g = 0; while (hi >= lo) { if (g == fileGroups.size()) { group = new LinkedList<Pair<SnapshotFileInfo, Long>>(); fileGroups.add(group); } else { group = fileGroups.get(g); } Pair<SnapshotFileInfo, Long> fileInfo = files.get(hi--); sizeGroups[g] += fileInfo.getSecond(); group.add(fileInfo); g += dir; if (g == ngroups) { dir = -1; g = ngroups - 1; } else if (g < 0) { dir = 1; g = 0; } } if (LOG.isDebugEnabled()) { for (int i = 0; i < sizeGroups.length; ++i) { LOG.debug("export split=" + i + " size=" + StringUtils.humanReadableInt(sizeGroups[i])); } } return fileGroups; } }
ExportSnapshot extends Configured implements Tool { static List<List<Pair<SnapshotFileInfo, Long>>> getBalancedSplits( final List<Pair<SnapshotFileInfo, Long>> files, final int ngroups) { Collections.sort(files, new Comparator<Pair<SnapshotFileInfo, Long>>() { public int compare(Pair<SnapshotFileInfo, Long> a, Pair<SnapshotFileInfo, Long> b) { long r = a.getSecond() - b.getSecond(); return (r < 0) ? -1 : ((r > 0) ? 1 : 0); } }); List<List<Pair<SnapshotFileInfo, Long>>> fileGroups = new LinkedList<List<Pair<SnapshotFileInfo, Long>>>(); long[] sizeGroups = new long[ngroups]; int hi = files.size() - 1; int lo = 0; List<Pair<SnapshotFileInfo, Long>> group; int dir = 1; int g = 0; while (hi >= lo) { if (g == fileGroups.size()) { group = new LinkedList<Pair<SnapshotFileInfo, Long>>(); fileGroups.add(group); } else { group = fileGroups.get(g); } Pair<SnapshotFileInfo, Long> fileInfo = files.get(hi--); sizeGroups[g] += fileInfo.getSecond(); group.add(fileInfo); g += dir; if (g == ngroups) { dir = -1; g = ngroups - 1; } else if (g < 0) { dir = 1; g = 0; } } if (LOG.isDebugEnabled()) { for (int i = 0; i < sizeGroups.length; ++i) { LOG.debug("export split=" + i + " size=" + StringUtils.humanReadableInt(sizeGroups[i])); } } return fileGroups; }  }
ExportSnapshot extends Configured implements Tool { static List<List<Pair<SnapshotFileInfo, Long>>> getBalancedSplits( final List<Pair<SnapshotFileInfo, Long>> files, final int ngroups) { Collections.sort(files, new Comparator<Pair<SnapshotFileInfo, Long>>() { public int compare(Pair<SnapshotFileInfo, Long> a, Pair<SnapshotFileInfo, Long> b) { long r = a.getSecond() - b.getSecond(); return (r < 0) ? -1 : ((r > 0) ? 1 : 0); } }); List<List<Pair<SnapshotFileInfo, Long>>> fileGroups = new LinkedList<List<Pair<SnapshotFileInfo, Long>>>(); long[] sizeGroups = new long[ngroups]; int hi = files.size() - 1; int lo = 0; List<Pair<SnapshotFileInfo, Long>> group; int dir = 1; int g = 0; while (hi >= lo) { if (g == fileGroups.size()) { group = new LinkedList<Pair<SnapshotFileInfo, Long>>(); fileGroups.add(group); } else { group = fileGroups.get(g); } Pair<SnapshotFileInfo, Long> fileInfo = files.get(hi--); sizeGroups[g] += fileInfo.getSecond(); group.add(fileInfo); g += dir; if (g == ngroups) { dir = -1; g = ngroups - 1; } else if (g < 0) { dir = 1; g = 0; } } if (LOG.isDebugEnabled()) { for (int i = 0; i < sizeGroups.length; ++i) { LOG.debug("export split=" + i + " size=" + StringUtils.humanReadableInt(sizeGroups[i])); } } return fileGroups; }  @Override int run(String[] args); static void main(String[] args); }
ExportSnapshot extends Configured implements Tool { static List<List<Pair<SnapshotFileInfo, Long>>> getBalancedSplits( final List<Pair<SnapshotFileInfo, Long>> files, final int ngroups) { Collections.sort(files, new Comparator<Pair<SnapshotFileInfo, Long>>() { public int compare(Pair<SnapshotFileInfo, Long> a, Pair<SnapshotFileInfo, Long> b) { long r = a.getSecond() - b.getSecond(); return (r < 0) ? -1 : ((r > 0) ? 1 : 0); } }); List<List<Pair<SnapshotFileInfo, Long>>> fileGroups = new LinkedList<List<Pair<SnapshotFileInfo, Long>>>(); long[] sizeGroups = new long[ngroups]; int hi = files.size() - 1; int lo = 0; List<Pair<SnapshotFileInfo, Long>> group; int dir = 1; int g = 0; while (hi >= lo) { if (g == fileGroups.size()) { group = new LinkedList<Pair<SnapshotFileInfo, Long>>(); fileGroups.add(group); } else { group = fileGroups.get(g); } Pair<SnapshotFileInfo, Long> fileInfo = files.get(hi--); sizeGroups[g] += fileInfo.getSecond(); group.add(fileInfo); g += dir; if (g == ngroups) { dir = -1; g = ngroups - 1; } else if (g < 0) { dir = 1; g = 0; } } if (LOG.isDebugEnabled()) { for (int i = 0; i < sizeGroups.length; ++i) { LOG.debug("export split=" + i + " size=" + StringUtils.humanReadableInt(sizeGroups[i])); } } return fileGroups; }  @Override int run(String[] args); static void main(String[] args);  }