@Test public void testSpilling() throws InterruptedException { Collection<ExternalIterable<Integer>> parts = tools.spillAndSortParts( IntStream.range(0, 1003) .boxed() .sorted(Comparator.reverseOrder()) .collect(Collectors.toList()), Integer::compare); assertEquals(11, parts.size()); assertEquals(11, spillFiles.getCreatedFiles().size()); }
@Override public <T> Collection<ExternalIterable<T>> spillAndSortParts( Iterable<T> what, Comparator<T> comparator) throws InterruptedException { List<ExternalIterable<T>> ret = new ArrayList<>(); List<T> sortList = new ArrayList<>(numSpillRecords); for (T e : what) { if (sortList.size() == numSpillRecords) { ret.add(externalize(sortList.stream().sorted(comparator))); LOG.debug("Successfully externalized {} records", sortList.size()); sortList.clear(); } sortList.add(e); if (Thread.currentThread().isInterrupted()) { throw new InterruptedException(); } } if (!sortList.isEmpty()) { ret.add(externalize(sortList.stream().sorted(comparator))); } return ret; }
GenericSpillTools implements SpillTools { @Override public <T> Collection<ExternalIterable<T>> spillAndSortParts( Iterable<T> what, Comparator<T> comparator) throws InterruptedException { List<ExternalIterable<T>> ret = new ArrayList<>(); List<T> sortList = new ArrayList<>(numSpillRecords); for (T e : what) { if (sortList.size() == numSpillRecords) { ret.add(externalize(sortList.stream().sorted(comparator))); LOG.debug("Successfully externalized {} records", sortList.size()); sortList.clear(); } sortList.add(e); if (Thread.currentThread().isInterrupted()) { throw new InterruptedException(); } } if (!sortList.isEmpty()) { ret.add(externalize(sortList.stream().sorted(comparator))); } return ret; } }
GenericSpillTools implements SpillTools { @Override public <T> Collection<ExternalIterable<T>> spillAndSortParts( Iterable<T> what, Comparator<T> comparator) throws InterruptedException { List<ExternalIterable<T>> ret = new ArrayList<>(); List<T> sortList = new ArrayList<>(numSpillRecords); for (T e : what) { if (sortList.size() == numSpillRecords) { ret.add(externalize(sortList.stream().sorted(comparator))); LOG.debug("Successfully externalized {} records", sortList.size()); sortList.clear(); } sortList.add(e); if (Thread.currentThread().isInterrupted()) { throw new InterruptedException(); } } if (!sortList.isEmpty()) { ret.add(externalize(sortList.stream().sorted(comparator))); } return ret; } GenericSpillTools(
      SerializerFactory serializer,
      SpillFileFactory spillFactory,
      Settings settings);  GenericSpillTools(
      SerializerFactory serializer,
      Settings settings); @VisibleForTesting  GenericSpillTools(
      SerializerFactory serializer,
      SpillFileFactory spillFactory,
      int spillRecords); }
GenericSpillTools implements SpillTools { @Override public <T> Collection<ExternalIterable<T>> spillAndSortParts( Iterable<T> what, Comparator<T> comparator) throws InterruptedException { List<ExternalIterable<T>> ret = new ArrayList<>(); List<T> sortList = new ArrayList<>(numSpillRecords); for (T e : what) { if (sortList.size() == numSpillRecords) { ret.add(externalize(sortList.stream().sorted(comparator))); LOG.debug("Successfully externalized {} records", sortList.size()); sortList.clear(); } sortList.add(e); if (Thread.currentThread().isInterrupted()) { throw new InterruptedException(); } } if (!sortList.isEmpty()) { ret.add(externalize(sortList.stream().sorted(comparator))); } return ret; } GenericSpillTools(
      SerializerFactory serializer,
      SpillFileFactory spillFactory,
      Settings settings);  GenericSpillTools(
      SerializerFactory serializer,
      Settings settings); @VisibleForTesting  GenericSpillTools(
      SerializerFactory serializer,
      SpillFileFactory spillFactory,
      int spillRecords); @Override ExternalIterable<T> externalize(Iterable<T> what); @Override Collection<ExternalIterable<T>> spillAndSortParts(
      Iterable<T> what, Comparator<T> comparator); }
GenericSpillTools implements SpillTools { @Override public <T> Collection<ExternalIterable<T>> spillAndSortParts( Iterable<T> what, Comparator<T> comparator) throws InterruptedException { List<ExternalIterable<T>> ret = new ArrayList<>(); List<T> sortList = new ArrayList<>(numSpillRecords); for (T e : what) { if (sortList.size() == numSpillRecords) { ret.add(externalize(sortList.stream().sorted(comparator))); LOG.debug("Successfully externalized {} records", sortList.size()); sortList.clear(); } sortList.add(e); if (Thread.currentThread().isInterrupted()) { throw new InterruptedException(); } } if (!sortList.isEmpty()) { ret.add(externalize(sortList.stream().sorted(comparator))); } return ret; } GenericSpillTools(
      SerializerFactory serializer,
      SpillFileFactory spillFactory,
      Settings settings);  GenericSpillTools(
      SerializerFactory serializer,
      Settings settings); @VisibleForTesting  GenericSpillTools(
      SerializerFactory serializer,
      SpillFileFactory spillFactory,
      int spillRecords); @Override ExternalIterable<T> externalize(Iterable<T> what); @Override Collection<ExternalIterable<T>> spillAndSortParts(
      Iterable<T> what, Comparator<T> comparator);  }