@ file : Suppress (" NOTH ING _ TO _ INLINE ", Ġ" Has Platform Type ") Ċ Ċ package Ġru . in for ion . lab 4 0 3 . common . spark Ċ Ċ import Ġorg . apache . spark . Task Context Ċ import Ġorg . apache . spark . api . java . Java RDD Ċ import Ġorg . apache . spark . api . java . Java RDD Like Ċ import Ġru . in for ion . lab 4 0 3 . common . concurrent . launch Ċ import Ġru . in for ion . lab 4 0 3 . common . extensions . INT _ MAX Ċ import Ġjava . util . concurrent . Linked BlockingQueue Ċ Ċ /** Ċ Ġ* ĠChanges Ġparameters Ġorder Ġto Ġcall Ġwith Ġ{ Ġ... Ġ} Ġnotation Ċ Ġ* Ċ Ġ* Ġ@ see ĠJava RDD . map Partitions Indexed Ċ Ġ*/ Ċ inline Ġfun Ġ< T , ĠR > ĠJava RDD < T >. map Partitions Indexed ( ĊĠĠĠ Ġpre serves Partition ing : ĠBoolean Ġ= Ġfalse , ĊĠĠĠ Ġno inline Ġaction : Ġ( Int , ĠIterator < T >) Ġ-> ĠIterator < R > Ċ ): ĠJava RDD < R > Ġ= Ġmap Partitions With Index ( action , Ġpre serves Partition ing ) ĊĊ Ċ inline Ġfun Ġ< T > ĠJava RDD < T >. forEach Partitions Indexed ( no inline Ġaction : Ġ( Int , ĠIterator < T >) Ġ-> ĠUnit ) Ġ= ĊĠĠĠ Ġforeach Partition Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġindex Ġ= ĠTask Context . get Partition Id () ĊĠĠĠĠĠĠĠ Ġaction ( index , Ġit ) ĊĠĠĠ Ġ} Ċ Ċ /** Ċ Ġ* ĠCollect s Ġitems Ġfrom ĠR DD Ġin Ġseparated Ġthread Ġasynchronous Ġand Ġreturn Ġiterator Ġto Ġcollected Ġitems Ċ Ġ* Ċ Ġ* Ġ@ param Ġcapacity Ġmaximum Ġcapacity Ġof Ġcache Ġqueue Ċ Ġ* Ċ Ġ* Ġ@ return Ġiterator Ġto Ġcollected Ġitems Ċ Ġ*/ Ċ inline Ġfun Ġ< T , ĠU Ġ: ĠJava RDD Like < T , ĠU >> ĠJava RDD Like < T , ĠU >. async Collect ( capacity : ĠInt Ġ= ĠINT _ MAX ) Ġ= Ġobject Ġ: ĠIterator < T > Ġ{ ĊĠĠĠ Ġprivate Ġval Ġempty Marker Ġ= Ġthis ĊĊĠĠĠ Ġprivate Ġvar Ġnext : ĠAny ? Ġ= Ġempty Marker ĊĊĠĠĠ Ġprivate Ġval Ġqueue Ġ= ĠLinked BlockingQueue < Any ? >( capacity ) ĊĊĠĠĠ Ġoverride Ġfun Ġhas Next (): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġnext Ġ= Ġqueue . take () ĊĠĠĠĠĠĠĠ Ġreturn Ġnext Ġ!== Ġempty Marker ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġnext (): ĠT Ġ{ ĊĠĠĠĠĠĠĠ Ġcheck ( next Ġ!== Ġempty Marker ) Ġ{ Ġ" Iterator Ġhas Ġno Ġnext Ġelement !" Ġ} ĊĠĠĠĠĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠĠĠĠĠ Ġreturn Ġnext Ġas ĠT ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġvar Ġindex Ġ= Ġ 0 ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġlaunch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġiterator Ġ= Ġto Local Iterator () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġiterator . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprintln (" Add Ġto Ġqueue : Ġ${ index ++ }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġqueue . add ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprintln (" Finished ") ĊĠĠĠĠĠĠĠ Ġ}. invoke On Completion Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprintln (" Add Ġempty Ġmarker ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġqueue . add ( empty Marker ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit ?. printStackTrace () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ inline Ġfun Ġ< T , ĠU Ġ: ĠJava RDD Like < T , ĠU >> ĠJava RDD Like < T , ĠU >. collect Partition ( index : ĠInt ): ĠList < T > Ġ= ĊĠĠĠ Ġcollect Partitions ( int ArrayOf ( index )). first () Ċ Ċ inline Ġfun Ġ< T > ĠJava RDD < T >. random ( fraction : ĠDouble ) Ġ= Ġsample ( false , Ġfraction , ĠSystem . currentTimeMillis ()) Ċ Ċ // Ġsomehow Ġget Ġsession Ġfrom Ġr dd Ġand Ġget Ġexec utors Ġcount Ċ inline Ġfun Ġ< T , ĠS > ĠJava RDD < T >. sort ( ascending : ĠBoolean Ġ= Ġfalse , Ġpartitions : ĠInt Ġ= Ġ 1 , Ġno inline Ġselector : Ġ( T ) Ġ-> ĠS ): ĠJava RDD < T > Ġ= ĊĠĠĠ ĠsortBy ( selector , Ġascending , Ġpartitions ) Ċ