/* Ċ Ġ* ĠCopyright Ġ 2 0 2 0 ĠAlex ĠCh erm en in Ċ Ġ* Ċ Ġ* ĠLicensed Ġunder Ġthe ĠApache ĠLicense , ĠVersion Ġ 2 . 0 Ġ( the Ġ" License "); Ċ Ġ* Ġyou Ġmay Ġnot Ġuse Ġthis Ġfile Ġexcept Ġin Ġcompliance Ġwith Ġthe ĠLicense . Ċ Ġ* ĠYou Ġmay Ġobtain Ġa Ġcopy Ġof Ġthe ĠLicense Ġat Ċ Ġ* Ċ Ġ* ĠĠĠĠ Ġhttp :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ Ġ* Ċ Ġ* ĠUnless Ġrequired Ġby Ġapplicable Ġlaw Ġor Ġagreed Ġto Ġin Ġwriting , Ġsoftware Ċ Ġ* Ġdistributed Ġunder Ġthe ĠLicense Ġis Ġdistributed Ġon Ġan Ġ" AS ĠIS " ĠBASIS , Ċ Ġ* ĠWITHOUT ĠWARRANTIES ĠOR ĠCONDITIONS ĠOF ĠANY ĠKIND , Ġeither Ġexpress Ġor Ġimplied . Ċ Ġ* ĠSee Ġthe ĠLicense Ġfor Ġthe Ġspecific Ġlanguage Ġgoverning Ġpermissions Ġand Ċ Ġ* Ġlimitations Ġunder Ġthe ĠLicense . Ċ Ġ*/ Ċ Ċ package Ġru . ch erm en in . k io . it Ċ Ċ import Ġjava . io . Serializable Ċ import Ġorg . junit . Test Ċ import Ġru . ch erm en in . k io . functions .* Ċ import Ġru . ch erm en in . k io . test .* Ċ Ċ class ĠK io Extensions IntegrationTest Ġ: ĠK io Pipeline Test () Ġ{ ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Pipeline () Ġ{ ĊĊĠĠĠĠĠĠĠ Ġ// ĠCreate Ġinputs ĊĠĠĠĠĠĠĠ Ġval Ġinput 1 Ġ= Ġk io . par alle lize ( listOf ( 1 , Ġ 2 , Ġ 3 )) ĊĠĠĠĠĠĠĠ Ġval Ġinput 2 Ġ= Ġk io . par alle lize ( listOf ( 4 , Ġ 5 , Ġ 6 )) ĊĊĠĠĠĠĠĠĠ Ġ// ĠConfigure Ġpipelines ĊĠĠĠĠĠĠĠ Ġval Ġresults Ġ= Ġinput 1 . union ( input 2 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit Ġ% Ġ 2 Ġ> Ġ 0 Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġnum Ġ-> ĠarrayOf Null s < String >( num ). map Ġ{ ĠString ( char ArrayOf (( num Ġ+ Ġ 4 2 ). to Char ())) Ġ} Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĠString ( char ArrayOf ( it [ 0 ])) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. key By Ġ{ Ġit [ 0 ]. toInt () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. group ByKey () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. to Pair () ĊĊĠĠĠĠĠĠĠ Ġ// ĠAssert Ġresults ĊĠĠĠĠĠĠĠ Ġresults . that (). contains In Any Order ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠPair ( 4 3 , ĠarrayOf ("+ "). as Iterable ()), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠPair ( 4 5 , ĠarrayOf ("- ", Ġ"-", Ġ"- "). as Iterable ()), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠPair ( 4 7 , ĠarrayOf ("/", Ġ"/", Ġ"/", Ġ"/", Ġ"/ "). as Iterable ()) ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġ// ĠExecute Ġpipeline ĊĠĠĠĠĠĠĠ Ġk io . execute (). wait Until Done () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Sql () Ġ{ ĊĠĠĠĠĠĠĠ Ġdata Ġclass ĠPerson ( val Ġid : ĠInt , Ġval Ġname : ĠString , Ġval Ġage : ĠInt ) Ġ: ĠSerializable ĊĊĠĠĠĠĠĠĠ Ġk io ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. par alle lize ( listOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPerson ( 0 , Ġ" John ", Ġ 2 6 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPerson ( 1 , Ġ" P eter ", Ġ 1 3 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPerson ( 2 , Ġ" Tim ", Ġ 6 4 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPerson ( 3 , Ġ" J ane ", Ġ 3 1 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPerson ( 4 , Ġ" Bill ", Ġ 1 7 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. to Rows () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. sql (" SELECT Ġid ĠFROM ĠP COLLECTION ĠWHERE Ġage Ġ>= Ġ 1 8 ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . getValue < Int >(" id ") Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. that (). contains In Any Order ( 0 , Ġ 2 , Ġ 3 ) ĊĊĠĠĠĠĠĠĠ Ġk io . execute (). wait Until Done () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Join () Ġ{ ĊĊĠĠĠĠĠĠĠ Ġdata Ġclass ĠPerson ( val Ġid : ĠInt , Ġval Ġname : ĠString , Ġval Ġage : ĠInt , Ġval Ġcity Id : ĠInt ) Ġ: ĠSerializable ĊĊĠĠĠĠĠĠĠ Ġdata Ġclass ĠCity ( val Ġid : ĠInt , Ġval Ġname : ĠString ) Ġ: ĠSerializable ĊĊĠĠĠĠĠĠĠ Ġval Ġpersons Ġ= Ġk io ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. par alle lize ( listOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPerson ( 0 , Ġ" John ", Ġ 2 6 , Ġ 2 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPerson ( 1 , Ġ" P eter ", Ġ 1 3 , Ġ 0 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPerson ( 2 , Ġ" Tim ", Ġ 6 4 , Ġ 1 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPerson ( 3 , Ġ" J ane ", Ġ 3 1 , Ġ 1 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPerson ( 4 , Ġ" Bill ", Ġ 1 7 , Ġ 2 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĊĠĠĠĠĠĠĠ Ġval Ġcities Ġ= Ġk io ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. par alle lize ( listOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCity ( 0 , Ġ" M os cow "), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCity ( 1 , Ġ" Par is "), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCity ( 2 , Ġ" L ondon "), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠCity ( 4 , Ġ" To ky o ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ)) ĊĊĠĠĠĠĠĠĠ Ġwith ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpersons . to Rows () Ġto Ġ" persons ", ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcities . to Rows () Ġto Ġ" cities " ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. sql (""" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠSELECT Ġp . name ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠFROM Ġpersons Ġp ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠJOIN Ġcities Ġc ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠON Ġp . city Id Ġ= Ġc . id ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠWHERE Ġc . name Ġ= Ġ' Par is ' ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"" ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . getValue < String >(" name ") Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. that (). contains In Any Order (" Tim ", Ġ" J ane ") ĊĊĠĠĠĠĠĠĠ Ġk io . execute (). wait Until Done () ĊĠĠĠ Ġ} Ċ } Ċ