package Ġmicro . apps . k beam Ċ Ċ import Ġjava . io . Serializable Ċ import Ġkotlin . test . Test Ċ import Ġmicro . apps . k beam . io . write Text Ċ import Ġorg . apache . beam . sdk . io . Compression Ċ import Ġorg . apache . beam . sdk . options . Value Provider . Static Value Provider Ċ import Ġorg . apache . beam . sdk . testing . Test Pipeline Ċ import Ġorg . apache . beam . sdk . transforms . Create Ċ import Ġorg . apache . beam . sdk . values . KV Ċ import Ġorg . junit . Rule Ċ Ċ class ĠK Beam DSL Test Ġ: ĠSerializable Ġ{ ĊĊĠĠĠ Ġ@ Rule ĊĠĠĠ Ġ@ Transient ĊĠĠĠ Ġ@ Jvm Field ĊĠĠĠ Ġval Ġpipeline Ġ= ĠTest Pipeline . create () ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun ĠShould Run Co Group ByKey DSL Test () Ġ{ ĊĊĠĠĠĠĠĠĠ Ġval Ġlist 1 Ġ= Ġ( 1 .. 2 0 ). map Ġ{ ĠK V . of (" Key _${ it Ġ% Ġ 1 0 })", Ġit ) Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġlist 2 Ġ= Ġ( 0 .. 9 ). flatMap Ġ{ Ġ(' a ' .. ' z '). map Ġ{ ĠK V . of (" Key _$ it )", Ġ"$ it ") Ġ} Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġplist 1 Ġ= Ġpipeline . apply (" Create ĠList 1 ", ĠCreate . of ( list 1 )) ĊĠĠĠĠĠĠĠ Ġval Ġplist 2 Ġ= Ġpipeline . apply (" Create ĠList 2 ", ĠCreate . of ( list 2 )) ĊĊĠĠĠĠĠĠĠ Ġval Ġgroup Ġ= Ġpipeline . co Group ByKey ( plist 1 , Ġplist 2 ) Ġ{ Ġkey , Ġleft , Ġright Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ ĠlistOf ("$ key Ġ: Ġ$ left Ġ$ right ") ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġgroup . par Do < String , ĠVoid > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprintln ( element ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġgroup . write Text Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpath Ġ= ĠStatic Value Provider . of (" build / output / test ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcompression Ġ= ĠCompression . GZIP ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnum Shards Ġ= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġheader Ġ= Ġ"# Header " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfooter Ġ= Ġ"# Footer " ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġpipeline . run (). wait Until Finish () ĊĠĠĠ Ġ} Ċ } Ċ