package Ġno . ok . orig o . dat apl atform . json transformer Ċ Ċ import Ġcom . amazonaws . services . lambda . runtime . Context Ċ import Ġcom . fasterxml . jackson . databind . JsonNode Ċ import Ġcom . fasterxml . jackson . module . kotlin . jackson ObjectMapper Ċ import Ġcom . fasterxml . jackson . module . kotlin . readValue Ċ import Ġcom . sch ib st ed . s pt . data . js lt . Expression Ċ import Ġcom . sch ib st ed . s pt . data . js lt . Parser Ċ import Ġno . ok . orig o . dat apl atform . commons . lambda . Dat apl atform Logging Handler Ċ import Ġno . ok . orig o . dat apl atform . commons . pipeline . config . Config Ċ import Ġno . ok . orig o . dat apl atform . commons . pipeline . config . Step Data Ċ import Ġsoftware . amazon . aw ssdk . core . sync . RequestBody Ċ import Ġsoftware . amazon . aw ssdk . services . s 3 . S 3 Client Ċ import Ġjava . io . InputStream Ċ import Ġjava . io . OutputStream Ċ Ċ class ĠJson 2 Json Handler ( private Ġval Ġs 3 : ĠS 3 Client Ġ= ĠS 3 Client . create ()) Ġ: ĠDat apl atform Logging Handler () Ġ{ ĊĊĠĠĠ Ġval Ġbucket Ġ= ĠSystem . getenv (" BUCKET _ NAME ") Ġ?: ĠSystem . getProperty (" BUCKET _ NAME ") ĊĠĠĠ Ġval Ġmapper Ġ= Ġj ack son ObjectMapper () ĊĊĠĠĠ Ġoverride Ġfun Ġhandle Request With Logging ( input : ĠInputStream , Ġoutput : ĠOutputStream , Ġcontext : ĠContext ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġconfig Ġ= Ġmapper . readValue < Config >( input ) ĊĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġhandle Request ( config ) ĊĠĠĠĠĠĠĠ Ġmapper . writeValue ( output , Ġresult ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġhandle Request ( config : ĠConfig ): ĠStep Data Ġ{ ĊĠĠĠĠĠĠĠ Ġlog Add (" execution _ name " Ġto Ġconfig . execution Name , Ġ" task " Ġto Ġconfig . task ) ĊĊĠĠĠĠĠĠĠ Ġval Ġtask Config Ġ= Ġconfig . get Task Config () ĊĠĠĠĠĠĠĠ Ġval Ġj slt Ġ= ĠParser . compile String ( task Config . get (" js lt "). text Value ()) ĊĊĠĠĠĠĠĠĠ Ġval Ġstep Data Ġ= Ġconfig . payload . step Data ĊĊĠĠĠĠĠĠĠ Ġreturn Ġif Ġ( step Data . s 3 Input Prefixes Ġ!= Ġnull Ġ&& Ġstep Data . s 3 Input Prefixes !!. isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġhandle S 3 Event ( config , Ġj slt ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( step Data . input Events Ġ!= Ġnull Ġ&& Ġstep Data . input Events !!. isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstep Data . input Events Ġ= Ġstep Data . input Events !!. map ( js lt :: apply ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstep Data ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠRuntimeException (" No Ġdata Ġto Ġtransform ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġhandle S 3 Event ( config : ĠConfig , Ġj slt : ĠExpression ): ĠStep Data Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġoutput Dataset Ġ= Ġconfig . payload . output Dataset ĊĠĠĠĠĠĠĠ Ġval Ġstep Data Ġ= Ġconfig . payload . step Data ĊĊĠĠĠĠĠĠĠ Ġval Ġnum Inputs Ġ= Ġstep Data . s 3 Input Prefixes !!. size ĊĠĠĠĠĠĠĠ Ġif Ġ( num Inputs Ġ> Ġ 1 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠToo Many Datasets ( num Inputs ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġ(_, Ġs 3 Input Prefix ) Ġ= Ġstep Data . s 3 Input Prefixes !!. iterator (). next () ĊĠĠĠĠĠĠĠ Ġlog Add (" s 3 _ input _ prefix " Ġto Ġs 3 Input Prefix ) ĊĠĠĠĠĠĠĠ Ġval Ġs 3 Objects Ġ= Ġlist Objects On Prefix ( s 3 Input Prefix ) ĊĠĠĠĠĠĠĠ Ġlog Add (" s 3 _ input _ keys " Ġto Ġs 3 Objects . map Ġ{ Ġit . key () Ġ}) ĊĠĠĠĠĠĠĠ Ġval Ġoutput Prefix Ġ= Ġconfig . get Intermediate Prefix () ĊĠĠĠĠĠĠĠ Ġlog Add (" s 3 _ output _ prefix " Ġto Ġoutput Prefix ) ĊĠĠĠĠĠĠĠ Ġs 3 Objects . forEach Ġ{ Ġs 3 Object Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġinput Key Ġ= Ġs 3 Object . key () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ( filename , Ġextension ) Ġ= Ġs 3 Object . partition Extension () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġoutput Key Ġ= Ġ"$ output Prefix $ filename .$ extension " ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġinput Json Ġ= Ġread From S 3 ( S 3 Object ( bucket , Ġinput Key )) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġoutput Json Ġ= Ġj slt . apply ( input Json ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwrite ToS 3 ( S 3 Object ( bucket , Ġoutput Key ), Ġoutput Json ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġstep Data . s 3 Input Prefixes Ġ= Ġmap Of ( output Dataset . id Ġto Ġoutput Prefix ) ĊĠĠĠĠĠĠĠ Ġreturn Ġstep Data ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġlist Objects On Prefix ( prefix : ĠString ): ĠList < software . amazon . aw ssdk . services . s 3 . model . S 3 Object > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġs 3 . list Objects V 2 Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġit . bucket ( bucket ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. prefix ( prefix ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġresult . contents (). toList () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġread From S 3 ( s 3 Object : ĠS 3 Object ): ĠJsonNode Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġs 3 . getObject Ġ{ Ġr Ġ-> Ġr . bucket ( s 3 Object . bucket ). key ( s 3 Object . key ) Ġ}. use Ġ{ ĠinputStream Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmapper . read Tree ( inputStream ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġwrite ToS 3 ( s 3 Object : ĠS 3 Object , Ġjson : ĠJsonNode ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġjson String Ġ= Ġmapper . writeValue AsString ( json ) ĊĠĠĠĠĠĠĠ Ġs 3 . put Object ({ Ġr Ġ-> Ġr . bucket ( s 3 Object . bucket ). key ( s 3 Object . key ) Ġ}, ĠRequest Body . fromString ( json String )) ĊĠĠĠ Ġ} Ċ } Ċ Ċ class ĠToo Many Datasets ( total Datasets : ĠInt ) Ġ: ĠException (" Too Ġmany Ġdatasets Ġin Ġinput Ġconfig : Ġ$ total Datasets ") Ċ