SortValues extends PTransform<
        PCollection<KV<PrimaryKeyT, Iterable<KV<SecondaryKeyT, ValueT>>>>,
        PCollection<KV<PrimaryKeyT, Iterable<KV<SecondaryKeyT, ValueT>>>>> { public static <PrimaryKeyT, SecondaryKeyT, ValueT> SortValues<PrimaryKeyT, SecondaryKeyT, ValueT> create( BufferedExternalSorter.Options sorterOptions) { return new SortValues<>(sorterOptions); } private  SortValues(BufferedExternalSorter.Options sorterOptions); static SortValues<PrimaryKeyT, SecondaryKeyT, ValueT> create(
          BufferedExternalSorter.Options sorterOptions); @Override PCollection<KV<PrimaryKeyT, Iterable<KV<SecondaryKeyT, ValueT>>>> expand(
      PCollection<KV<PrimaryKeyT, Iterable<KV<SecondaryKeyT, ValueT>>>> input);  }
@Test public void testSecondaryKeySorting() { PCollection<KV<String, KV<String, Integer>>> input = p.apply( Create.of( Arrays.asList( KV.of("key1", KV.of("secondaryKey2", 20)), KV.of("key2", KV.of("secondaryKey2", 200)), KV.of("key1", KV.of("secondaryKey3", 30)), KV.of("key1", KV.of("secondaryKey1", 10)), KV.of("key2", KV.of("secondaryKey1", 100))))); PCollection<KV<String, Iterable<KV<String, Integer>>>> grouped = input.apply(GroupByKey.create()); PCollection<KV<String, Iterable<KV<String, Integer>>>> groupedAndSorted = grouped.apply(SortValues.create(BufferedExternalSorter.options())); PAssert.that(groupedAndSorted) .satisfies(new AssertThatHasExpectedContentsForTestSecondaryKeySorting()); p.run(); }