JoinExamples { static PCollection<String> joinEvents( PCollection<TableRow> eventsTable, PCollection<TableRow> countryCodes) throws Exception { final TupleTag<String> eventInfoTag = new TupleTag<>(); final TupleTag<String> countryInfoTag = new TupleTag<>(); PCollection<KV<String, String>> eventInfo = eventsTable.apply(ParDo.of(new ExtractEventDataFn())); PCollection<KV<String, String>> countryInfo = countryCodes.apply(ParDo.of(new ExtractCountryInfoFn())); PCollection<KV<String, CoGbkResult>> kvpCollection = KeyedPCollectionTuple.of(eventInfoTag, eventInfo) .and(countryInfoTag, countryInfo) .apply(CoGroupByKey.create()); PCollection<KV<String, String>> finalResultCollection = kvpCollection.apply( "Process", ParDo.of( new DoFn<KV<String, CoGbkResult>, KV<String, String>>() { @ProcessElement public void processElement(ProcessContext c) { KV<String, CoGbkResult> e = c.element(); String countryCode = e.getKey(); String countryName = "none"; countryName = e.getValue().getOnly(countryInfoTag); for (String eventInfo : c.element().getValue().getAll(eventInfoTag)) { c.output( KV.of( countryCode, "Country name: " + countryName + ", Event info: " + eventInfo)); } } })); PCollection<String> formattedResults = finalResultCollection.apply( "Format", ParDo.of( new DoFn<KV<String, String>, String>() { @ProcessElement public void processElement(ProcessContext c) { String outputstring = "Country code: " + c.element().getKey() + ", " + c.element().getValue(); c.output(outputstring); } })); return formattedResults; }  static void main(String[] args);  }
@Test @Category(ValidatesRunner.class) public void testJoin() throws java.lang.Exception { PCollection<TableRow> input1 = p.apply("CreateEvent", Create.of(EVENT_ARRAY)); PCollection<TableRow> input2 = p.apply("CreateCC", Create.of(CC_ARRAY)); PCollection<String> output = JoinExamples.joinEvents(input1, input2); PAssert.that(output).containsInAnyOrder(JOINED_EVENTS); p.run().waitUntilFinish(); }