Top { private Top() { } private  Top(); static Combine.Globally<T, List<T>> of(int count, ComparatorT compareFn); static Combine.Globally<T, List<T>> smallest(int count); static Combine.Globally<T, List<T>> largest(int count); static TopCombineFn<T, Natural<T>> largestFn(int count); static TopCombineFn<Long, Natural<Long>> largestLongsFn(int count); static TopCombineFn<Integer, Natural<Integer>> largestIntsFn(int count); static TopCombineFn<Double, Natural<Double>> largestDoublesFn(int count); static TopCombineFn<T, Reversed<T>> smallestFn(int count); static TopCombineFn<Long, Reversed<Long>> smallestLongsFn(int count); static TopCombineFn<Integer, Reversed<Integer>> smallestIntsFn(int count); static TopCombineFn<Double, Reversed<Double>> smallestDoublesFn(int count); static PTransform<PCollection<KV<K, V>>, PCollection<KV<K, List<V>>>> perKey(
          int count, ComparatorT compareFn); static PTransform<PCollection<KV<K, V>>, PCollection<KV<K, List<V>>>> smallestPerKey(int count); static PerKey<K, V, List<V>> largestPerKey(int count);  }
@Test @Category(NeedsRunner.class) @SuppressWarnings("unchecked") public void testTop() { PCollection<String> input = p.apply(Create.of(Arrays.asList(COLLECTION)).withCoder(StringUtf8Coder.of())); PCollection<List<String>> top1 = input.apply(Top.of(1, new OrderByLength())); PCollection<List<String>> top2 = input.apply(Top.largest(2)); PCollection<List<String>> top3 = input.apply(Top.smallest(3)); PCollection<KV<String, Integer>> inputTable = createInputTable(p); PCollection<KV<String, List<Integer>>> largestPerKey = inputTable.apply(Top.largestPerKey(2)); PCollection<KV<String, List<Integer>>> smallestPerKey = inputTable.apply(Top.smallestPerKey(2)); PAssert.thatSingletonIterable(top1).containsInAnyOrder(Arrays.asList("bb")); PAssert.thatSingletonIterable(top2).containsInAnyOrder("z", "c"); PAssert.thatSingletonIterable(top3).containsInAnyOrder("a", "bb", "c"); PAssert.that(largestPerKey) .containsInAnyOrder(KV.of("a", Arrays.asList(3, 2)), KV.of("b", Arrays.asList(100, 10))); PAssert.that(smallestPerKey) .containsInAnyOrder(KV.of("a", Arrays.asList(1, 2)), KV.of("b", Arrays.asList(1, 10))); p.run(); }