package Ġet l . loader Ċ Ċ import Ġcom . google . api . services . bigquery . model . Table Reference Ċ import Ġcom . google . api . services . bigquery . model . TableRow Ċ import Ġorg . apache . beam . sdk . io . gcp . bigquery . Big Query IO Ċ import Ġorg . apache . beam . sdk . transforms . Filter Ċ import Ġorg . apache . beam . sdk . transforms . Map Elements Ċ import Ġorg . apache . beam . sdk . transforms . Process Function Ċ import Ġorg . apache . beam . sdk . values . TypeDescriptor Ċ import Ġorg . apache . beam . sdk . values . Type Descriptors Ċ Ċ type alias ĠFile Line Ġ= ĠString Ċ type alias ĠSp ited File Line Ġ= ĠList < String > Ċ Ċ object ĠPipeline Steps Ġ{ ĊĠĠĠ Ġprivate Ġconst Ġval Ġheaders Ġ= Ġ""" " No "," year "," month "," day "," hour "," PM 2 . 5 "," PM 1 0 "," SO 2 "," NO 2 "," CO "," O 3 "," TEMP "," PRES "," DE WP "," R AIN "," wd "," W SP M "," station """" ĊĊĠĠĠ Ġfun Ġfilter Header Row (): ĠFilter < String > Ġ= ĠFilter . by ( Process Function < String , ĠBoolean > Ġ{ Ġ! it . startsWith ( headers ) Ġ}) ĊĊĠĠĠ Ġfun ĠwriteTo Big Query ( result Table Spec : ĠTable Reference ): ĠBig Query IO . Write < TableRow > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠBig Query IO . write Table Rows () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. to ( result Table Spec ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Schema ( Record Schema . schema ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Create Disposition ( Big Query IO . Write . Create Disposition . CREATE _ IF _ NE EDED ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Write Disposition ( Big Query IO . Write . Write Disposition . WRITE _ APPEND ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġsplit Csv Line (): ĠMap Elements < File Line , ĠSp ited File Line > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠMap Elements . into ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠType Descriptors . lists ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠType Descriptor . of ( String :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. via ( Process Function < File Line , ĠSp ited File Line > Ġ{ Ġit . split (",") Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġmap Entity To TableRow (): ĠMap Elements < Record , ĠTable Row > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠMap Elements . into ( TypeDescriptor . of ( TableRow :: class . java )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. via ( Process Function Ġ{ ĠRecord Schema . to TableRow ( it ) Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġmap Elements To Entity (): ĠMap Elements < Sp ited File Line , ĠRecord > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠMap Elements . into ( TypeDescriptor . of ( Record :: class . java )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. via ( Process Function < Sp ited File Line , ĠRecord > Ġ{ Ġr Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRecord ( timestamp Ġ= ĠUtils . csv To LocalDateTime ( r . sub List ( 1 , Ġ 5 )), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpm 2 5 Ġ= Ġr . getDouble OrNull ( 5 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpm 1 0 Ġ= Ġr . getDouble OrNull ( 6 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġso 2 Ġ= Ġr . getDouble OrNull ( 7 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġno 2 Ġ= Ġr . getDouble OrNull ( 8 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġco Ġ= Ġr . getDouble OrNull ( 9 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġo 3 Ġ= Ġr . getDouble OrNull ( 1 0 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtemp Ġ= Ġr . getDouble OrNull ( 1 1 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpress Ġ= Ġr . getDouble OrNull ( 1 2 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġde wp Ġ= Ġr . getDouble OrNull ( 1 3 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrain Ġ= Ġr . getDouble OrNull ( 1 4 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwd Ġ= Ġr . get OrNull ( 1 5 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġw sp m Ġ= Ġr . getDouble OrNull ( 1 6 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstation Ġ= Ġr . get OrNull ( 1 7 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} Ċ } Ċ Ċ private Ġfun ĠList < String >. getDouble OrNull ( i : ĠInt ) Ġ= Ġget OrNull ( i )?. toDouble OrNull () Ċ