/* Ċ Ġ* ĠLicensed Ġto Ġthe ĠApache ĠSoftware ĠFoundation Ġ( ASF ) Ġunder Ġone Ċ Ġ* Ġor Ġmore Ġcontributor Ġlicense Ġagreements . Ġ ĠSee Ġthe ĠNOTICE Ġfile Ċ Ġ* Ġdistributed Ġwith Ġthis Ġwork Ġfor Ġadditional Ġinformation Ċ Ġ* Ġregarding Ġcopyright Ġownership . Ġ ĠThe ĠASF Ġlicenses Ġthis Ġfile Ċ Ġ* Ġto Ġyou Ġ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 Ġorg . apache . beam . examples . kotlin Ċ Ċ import Ġorg . apache . beam . examples . kotlin . common . Example Utils Ċ import Ġorg . apache . beam . sdk . Pipeline Ċ import Ġorg . apache . beam . sdk . io . Text IO Ċ import Ġorg . apache . beam . sdk . metrics . Metrics Ċ import Ġorg . apache . beam . sdk . options . Default Ċ import Ġorg . apache . beam . sdk . options . Description Ċ import Ġorg . apache . beam . sdk . options . Pipeline Options Ċ import Ġorg . apache . beam . sdk . options . Pipeline Options Factory Ċ import Ġorg . apache . beam . sdk . options . Validation . Required Ċ import Ġorg . apache . beam . sdk . transforms .* Ċ import Ġorg . apache . beam . sdk . values . KV Ċ import Ġorg . apache . beam . sdk . values . P Collection Ċ import Ġorg . apache . beam . sdk . values . P Done Ċ Ċ /** Ċ Ġ* ĠAn Ġexample Ġthat Ġcounts Ġwords Ġin ĠSh akes pe are Ġand Ġincludes ĠBe am Ġbest Ġpractices . Ċ Ġ* Ċ Ġ* Ċ Ġ* ĠThis Ġclass , Ġ[ Word Count ], Ġis Ġthe Ġsecond Ġin Ġa Ġseries Ġof Ġfour Ġsuccess ively Ġmore Ġdetailed Ċ Ġ* Ġ' word Ġcount ' Ġexamples . ĠYou Ġmay Ġfirst Ġwant Ġto Ġtake Ġa Ġlook Ġat Ġ[ Minimal Word Count ]. ĠAfter Ċ Ġ* Ġyou 've Ġlooked Ġat Ġthis Ġexample , Ġthen Ġsee Ġthe Ġ[ Debugging Word Count ] Ġpipeline , Ġfor Ġintroduction Ċ Ġ* Ġof Ġadditional Ġconcepts . Ċ Ġ* Ċ Ġ* Ċ Ġ* ĠFor Ġa Ġdetailed Ġwalk through Ġof Ġthis Ġexample , Ġsee Ġ[ Ċ Ġ* Ġhttps :// beam . apache . org / get - started / word count - example / Ġ]( https :// beam . apache . org / get - started / word count - example /) Ċ Ġ* Ċ Ġ* Ċ Ġ* ĠBasic Ġconcepts , Ġalso Ġin Ġthe ĠMin imal Word Count Ġexample : ĠReading Ġtext Ġfiles ; Ġcounting Ġa Ċ Ġ* ĠP Collection ; Ġwriting Ġto Ġtext Ġfiles Ċ Ġ* Ċ Ġ* Ċ Ġ* ĠNew ĠConcept s : Ċ Ġ* Ċ Ġ* Ġ< pre > Ċ Ġ* Ġ 1 . ĠExec uting Ġa ĠPipeline Ġboth Ġlocally Ġand Ġusing Ġthe Ġselected Ġrunner Ċ Ġ* Ġ 2 . ĠUsing ĠPar Do Ġwith Ġstatic ĠDo Fns Ġdefined Ġout - of - line Ċ Ġ* Ġ 3 . ĠBuilding Ġa Ġcomposite Ġtransform Ċ Ġ* Ġ 4 . ĠDef ining Ġyour Ġown Ġpipeline Ġoptions Ċ </ pre > Ġ* Ċ Ġ* Ċ Ġ* Ċ Ġ* ĠConcept Ġ# 1 : Ġyou Ġcan Ġexecute Ġthis Ġpipeline Ġeither Ġlocally Ġor Ġusing Ġby Ġselecting Ġanother Ġrunner . Ċ Ġ* ĠThese Ġare Ġnow Ġcommand - line Ġoptions Ġand Ġnot Ġhard - coded Ġas Ġthey Ġwere Ġin Ġthe ĠMin imal Word Count Ċ Ġ* Ġexample . Ċ Ġ* Ċ Ġ* Ċ Ġ* ĠTo Ġchange Ġthe Ġrunner , Ġspecify : Ċ Ġ* Ċ Ġ* Ġ< pre >` -- runner = YOUR _ SELECTED _ RUN NER Ċ ` </ pre > Ġ* Ċ Ġ* Ċ Ġ* Ċ Ġ* ĠTo Ġexecute Ġthis Ġpipeline , Ġspecify Ġa Ġlocal Ġoutput Ġfile Ġ( if Ġusing Ġthe Ġ` Direct Runner `) Ġor Ċ Ġ* Ġoutput Ġprefix Ġon Ġa Ġsupported Ġdistributed Ġfile Ġsystem . Ċ Ġ* Ċ Ġ* Ġ< pre >` -- output =[ YOUR _ LOCAL _ FILE Ġ| ĠYOUR _ OUTPUT _ PREFIX ] Ċ ` </ pre > Ġ* Ċ Ġ* Ċ Ġ* Ċ Ġ* ĠThe Ġinput Ġfile Ġdefaults Ġto Ġa Ġpublic Ġdata Ġset Ġcontaining Ġthe Ġtext Ġof Ġof ĠKing ĠLear , Ġby ĠW illiam Ċ Ġ* ĠSh akes pe are . ĠYou Ġcan Ġoverride Ġit Ġand Ġchoose Ġyour Ġown Ġinput Ġwith Ġ`-- inputFile `. Ċ Ġ*/ Ċ public Ġobject ĠWord Count Ġ{ ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠConcept Ġ# 2 : ĠYou Ġcan Ġmake Ġyour Ġpipeline Ġassembly Ġcode Ġless Ġverbose Ġby Ġdefining Ġyour ĠDo Fns ĊĠĠĠĠ Ġ* Ġstatic ally Ġout - of - line . ĠThis ĠDo Fn Ġtoken izes Ġlines Ġof Ġtext Ġinto Ġindividual Ġwords ; Ġwe Ġpass Ġit Ġto ĊĠĠĠĠ Ġ* Ġa ĠPar Do Ġin Ġthe Ġpipeline . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġpublic Ġclass ĠExtract Words Fn Ġ: ĠDo Fn < String , ĠString >() Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġempty Lines Ġ= ĠMetrics . counter ( Extract Words Fn :: class . java , Ġ" empty Lines ") ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġline Len Dist Ġ= ĠMetrics . distribution ( Extract Words Fn :: class . java , Ġ" line Len Dist ro ") ĊĊĠĠĠĠĠĠĠ Ġ@ Process Element ĊĠĠĠĠĠĠĠ Ġfun Ġprocess Element (@ Element Ġelement : ĠString , Ġreceiver : ĠDo Fn . Output Receiver < String >) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġline Len Dist . update ( element . length . to Long ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( element . trim (' Ġ'). isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġempty Lines . inc () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠSplit Ġthe Ġline Ġinto Ġwords . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġwords Ġ= Ġelement . split ( Example Utils . TOKEN IZER _ PATTERN . to Regex ()). to Typed Array () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠOutput Ġeach Ġword Ġencountered Ġinto Ġthe Ġoutput ĠP Collection . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfor Ġ( word Ġin Ġwords ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( word . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceiver . output ( word ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĠA ĠSimple Function Ġthat Ġconverts Ġa ĠWord Ġand ĠCount Ġinto Ġa Ġprint able Ġstring . Ġ Ġ*/ ĊĠĠĠ Ġpublic Ġclass ĠFormat As Text Fn Ġ: ĠSimple Function < KV < String , ĠLong >, ĠString >() Ġ{ ĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġapply ( input : ĠK V < String , ĠLong >) Ġ= Ġ"${ input . key } Ġ: Ġ${ input . value }" ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠA ĠP Transform Ġthat Ġconverts Ġa ĠP Collection Ġcontaining Ġlines Ġof Ġtext Ġinto Ġa ĠP Collection Ġof ĊĠĠĠĠ Ġ* Ġformatted Ġword Ġcounts . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠConcept Ġ# 3 : ĠThis Ġis Ġa Ġcustom Ġcomposite Ġtransform Ġthat Ġbundles Ġtwo Ġtransforms Ġ( Par Do Ġand ĊĠĠĠĠ Ġ* ĠCount ) Ġas Ġa Ġre usable ĠP Transform Ġsubclass . ĠUsing Ġcomposite Ġtransforms Ġallows Ġfor Ġeasy Ġreuse , ĊĠĠĠĠ Ġ* Ġmodular Ġtesting , Ġand Ġan Ġimproved Ġmonitoring Ġexperience . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġpublic Ġclass ĠCount Words Ġ: ĠP Transform < P Collection < String >, ĠP Collection < KV < String , ĠLong >> >() Ġ{ ĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġexpand ( lines : ĠP Collection < String >): ĠP Collection < KV < String , ĠLong >> Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠConvert Ġlines Ġof Ġtext Ġinto Ġindividual Ġwords . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġwords Ġ= Ġlines . apply ( Par Do . of ( Extract Words Fn ())) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠCount Ġthe Ġnumber Ġof Ġtimes Ġeach Ġword Ġoccurs . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġword Counts Ġ= Ġwords . apply ( Count . per Element ()) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġword Counts ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠOptions Ġsupported Ġby Ġ[ Word Count ]. ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠConcept Ġ# 4 : ĠDef ining Ġyour Ġown Ġconfiguration Ġoptions . ĠHere , Ġyou Ġcan Ġadd Ġyour Ġown Ġarguments Ġto ĊĠĠĠĠ Ġ* Ġbe Ġprocessed Ġby Ġthe Ġcommand - line Ġparser , Ġand Ġspecify Ġdefault Ġvalues Ġfor Ġthem . ĠYou Ġcan Ġthen ĊĠĠĠĠ Ġ* Ġaccess Ġthe Ġoptions Ġvalues Ġin Ġyour Ġpipeline Ġcode . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠIn herits Ġstandard Ġconfiguration Ġoptions . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġpublic Ġinterface ĠWord Count Options Ġ: ĠPipeline Options Ġ{ ĊĊĠĠĠĠĠĠĠ Ġ/** ĊĠĠĠĠĠĠĠĠ Ġ* ĠBy Ġdefault , Ġthis Ġexample Ġreads Ġfrom Ġa Ġpublic Ġdataset Ġcontaining Ġthe Ġtext Ġof ĠKing ĠLear . ĠSet ĊĠĠĠĠĠĠĠĠ Ġ* Ġthis Ġoption Ġto Ġchoose Ġa Ġdifferent Ġinput Ġfile Ġor Ġglob . ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġ@ get : Description (" Path Ġof Ġthe Ġfile Ġto Ġread Ġfrom ") ĊĠĠĠĠĠĠĠ Ġ@ get : Default . String (" gs :// apache - beam - samples / sh akes pe are / king lear . txt ") ĊĠĠĠĠĠĠĠ Ġvar Ġinput File : ĠString ĊĊĠĠĠĠĠĠĠ Ġ/** ĠSet Ġthis Ġrequired Ġoption Ġto Ġspecify Ġwhere Ġto Ġwrite Ġthe Ġoutput . Ġ Ġ*/ ĊĠĠĠĠĠĠĠ Ġ@ get : Description (" Path Ġof Ġthe Ġfile Ġto Ġwrite Ġto ") ĊĠĠĠĠĠĠĠ Ġ@ get : Required ĊĠĠĠĠĠĠĠ Ġvar Ġoutput : ĠString ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Jvm Static ĊĠĠĠ Ġfun Ġrun Word Count ( options : ĠWord Count Options ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġp Ġ= ĠPipeline . create ( options ) ĊĊĠĠĠĠĠĠĠ Ġ// ĠConcept s Ġ# 2 Ġand Ġ# 3 : ĠOur Ġpipeline Ġapplies Ġthe Ġcomposite ĠCount Words Ġtransform , Ġand Ġpasses Ġthe ĊĠĠĠĠĠĠĠ Ġ// Ġstatic ĠFormat As Text Fn () Ġto Ġthe ĠPar Do Ġtransform . ĊĠĠĠĠĠĠĠ Ġp . apply (" Read Lines ", ĠText IO . read (). from ( options . inputFile )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. apply ( Count Words ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. apply ( Map Elements . via ( Format As Text Fn ())) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. apply < P Done >(" Write Counts ", ĠText IO . write (). to ( options . output )) ĊĊĠĠĠĠĠĠĠ Ġp . run (). wait Until Finish () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Jvm Static ĊĠĠĠ Ġfun Ġmain ( args : ĠArray < String >) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġoptions Ġ= Ġ( Pipeline Options Factory . from Args (* args ). with Validation () Ġas ĠWord Count Options ) ĊĠĠĠĠĠĠĠ Ġrun Word Count ( options ) ĊĠĠĠ Ġ} Ċ } Ċ