Snippets { public static PCollection<String> coGroupByKeyTuple( TupleTag<String> emailsTag, TupleTag<String> phonesTag, PCollection<KV<String, String>> emails, PCollection<KV<String, String>> phones) { PCollection<KV<String, CoGbkResult>> results = KeyedPCollectionTuple.of(emailsTag, emails) .and(phonesTag, phones) .apply(CoGroupByKey.create()); PCollection<String> contactLines = results.apply( ParDo.of( new DoFn<KV<String, CoGbkResult>, String>() { @ProcessElement public void processElement(ProcessContext c) { KV<String, CoGbkResult> e = c.element(); String name = e.getKey(); Iterable<String> emailsIter = e.getValue().getAll(emailsTag); Iterable<String> phonesIter = e.getValue().getAll(phonesTag); String formattedResult = Snippets.formatCoGbkResults(name, emailsIter, phonesIter); c.output(formattedResult); } })); return contactLines; }  static void modelBigQueryIO(Pipeline p); static void modelBigQueryIO(
      Pipeline p, String writeProject, String writeDataset, String writeTable); static String formatCoGbkResults(
      String name, Iterable<String> emails, Iterable<String> phones); static PCollection<String> coGroupByKeyTuple(
      TupleTag<String> emailsTag,
      TupleTag<String> phonesTag,
      PCollection<KV<String, String>> emails,
      PCollection<KV<String, String>> phones); static void fileProcessPattern(); static void sideInputPatterns(); static void accessingValueProviderInfoAfterRunSnip1(String[] args); Collection<IntervalWindow> assignWindows(WindowFn.AssignContext c);  }
@Test public void testCoGroupByKeyTuple() throws IOException { final List<KV<String, String>> emailsList = Arrays.asList( KV.of("amy", "amy@example.com"), KV.of("carl", "carl@example.com"), KV.of("julia", "julia@example.com"), KV.of("carl", "carl@email.com")); final List<KV<String, String>> phonesList = Arrays.asList( KV.of("amy", "111-222-3333"), KV.of("james", "222-333-4444"), KV.of("amy", "333-444-5555"), KV.of("carl", "444-555-6666")); PCollection<KV<String, String>> emails = p.apply("CreateEmails", Create.of(emailsList)); PCollection<KV<String, String>> phones = p.apply("CreatePhones", Create.of(phonesList)); final TupleTag<String> emailsTag = new TupleTag<>(); final TupleTag<String> phonesTag = new TupleTag<>(); final List<KV<String, CoGbkResult>> expectedResults = Arrays.asList( KV.of( "amy", CoGbkResult.of(emailsTag, Arrays.asList("amy@example.com")) .and(phonesTag, Arrays.asList("111-222-3333", "333-444-5555"))), KV.of( "carl", CoGbkResult.of(emailsTag, Arrays.asList("carl@email.com", "carl@example.com")) .and(phonesTag, Arrays.asList("444-555-6666"))), KV.of( "james", CoGbkResult.of(emailsTag, Arrays.asList()) .and(phonesTag, Arrays.asList("222-333-4444"))), KV.of( "julia", CoGbkResult.of(emailsTag, Arrays.asList("julia@example.com")) .and(phonesTag, Arrays.asList()))); PCollection<String> actualFormattedResults = Snippets.coGroupByKeyTuple(emailsTag, phonesTag, emails, phones); final List<String> formattedResults = Arrays.asList( "amy; ['amy@example.com']; ['111-222-3333', '333-444-5555']", "carl; ['carl@email.com', 'carl@example.com']; ['444-555-6666']", "james; []; ['222-333-4444']", "julia; ['julia@example.com']; []"); List<String> expectedFormattedResultsList = new ArrayList<>(expectedResults.size()); for (KV<String, CoGbkResult> e : expectedResults) { String name = e.getKey(); Iterable<String> emailsIter = e.getValue().getAll(emailsTag); Iterable<String> phonesIter = e.getValue().getAll(phonesTag); String formattedResult = Snippets.formatCoGbkResults(name, emailsIter, phonesIter); expectedFormattedResultsList.add(formattedResult); } PCollection<String> expectedFormattedResultsPColl = p.apply(Create.of(expectedFormattedResultsList)); PAssert.that(expectedFormattedResultsPColl).containsInAnyOrder(formattedResults); PAssert.that(actualFormattedResults).containsInAnyOrder(formattedResults); p.run(); }