public static PCollection < String > coGroupByKeyTuple ( TupleTag < String > emailsTag , TupleTag < String > phonesTag , PCollection < KV < String , String > > emails , PCollection < KV < String , String > > phones ) { // [START CoGroupByKeyTuple] 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 ) ; } } ) ) ; // [END CoGroupByKeyTuple] return contactLines ; }
Using a CoGroupByKey transform .