static PCollection < String > joinEvents ( PCollection < TableRow > eventsTable , PCollection < TableRow > countryCodes ) throws Exception { final TupleTag < String > eventInfoTag = new TupleTag <> ( ) ; final TupleTag < String > countryInfoTag = new TupleTag <> ( ) ; // transform both input collections to tuple collections, where the keys are country // codes in both cases. PCollection < KV < String , String > > eventInfo = eventsTable . apply ( ParDo . of ( new ExtractEventDataFn ( ) ) ) ; PCollection < KV < String , String > > countryInfo = countryCodes . apply ( ParDo . of ( new ExtractCountryInfoFn ( ) ) ) ; // country code 'key' -> CGBKR (<event info>, <country name>) PCollection < KV < String , CoGbkResult > > kvpCollection = KeyedPCollectionTuple . of ( eventInfoTag , eventInfo ) . and ( countryInfoTag , countryInfo ) . apply ( CoGroupByKey . create ( ) ) ; // Process the CoGbkResult elements generated by the CoGroupByKey transform. // country code 'key' -> string of <event info>, <country name> 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 ) ) { // Generate a string that combines information from both collection values c . output ( KV . of ( countryCode , "Country name: " + countryName + ", Event info: " + eventInfo ) ) ; } } } ) ) ; // write to GCS 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 ; }
Join two collections using country code as the key .