CoGroupByKey extends PTransform<KeyedPCollectionTuple<K>, PCollection<KV<K, CoGbkResult>>> { private CoGroupByKey() {} private  CoGroupByKey(); static CoGroupByKey<K> create(); @Override PCollection<KV<K, CoGbkResult>> expand(KeyedPCollectionTuple<K> input);  }
@Test @Category({ValidatesRunner.class, UsesSideInputs.class}) public void testCoGroupByKey() { final TupleTag<String> namesTag = new TupleTag<>(); final TupleTag<String> addressesTag = new TupleTag<>(); final TupleTag<String> purchasesTag = new TupleTag<>(); PCollection<KV<Integer, CoGbkResult>> coGbkResults = buildPurchasesCoGbk(p, purchasesTag, addressesTag, namesTag); PAssert.thatMap(coGbkResults) .satisfies( results -> { CoGbkResult result1 = results.get(1); assertEquals("John Smith", result1.getOnly(namesTag)); assertThat(result1.getAll(purchasesTag), containsInAnyOrder("Shoes", "Book")); CoGbkResult result2 = results.get(2); assertEquals("Sally James", result2.getOnly(namesTag)); assertEquals("53 S. 3rd", result2.getOnly(addressesTag)); assertThat(result2.getAll(purchasesTag), containsInAnyOrder("Suit", "Boat")); CoGbkResult result3 = results.get(3); assertEquals("29 School Rd", "29 School Rd", result3.getOnly(addressesTag)); assertThat(result3.getAll(purchasesTag), containsInAnyOrder("Car", "House")); CoGbkResult result8 = results.get(8); assertEquals("Jeffery Spalding", result8.getOnly(namesTag)); assertEquals("6 Watling Rd", result8.getOnly(addressesTag)); assertThat(result8.getAll(purchasesTag), containsInAnyOrder("House", "Suit Case")); CoGbkResult result20 = results.get(20); assertEquals("Joan Lichtfield", result20.getOnly(namesTag)); assertEquals("3 W. Arizona", result20.getOnly(addressesTag)); assertEquals("383 Jackson Street", results.get(10).getOnly(addressesTag)); assertThat(results.get(4).getAll(purchasesTag), containsInAnyOrder("Suit")); assertThat(results.get(10).getAll(purchasesTag), containsInAnyOrder("Pens")); assertThat(results.get(11).getAll(purchasesTag), containsInAnyOrder("House")); assertThat(results.get(14).getAll(purchasesTag), containsInAnyOrder("Shoes")); return null; }); p.run(); }