Service]
    val registry = new Handler[IncrementalRegistryTableData]
    val podTimeZoneHelper = new Handler[PodDateTimeUtils]
    val billingPodStatusDao = new Handler[BillingPodStatusBDAO]
    val config = new Handler[PropertyManager]
    val paramFacade = new Handler[ConsumptionParameterFacade]
    val consumptionMethods = new Handler[ConsumptionMethods]
    val partitions = config.get.defaultPartitions()
    val appName = sc.appName
    val appId = sc.applicationId
    val now = new DateTime

    val extracted = ctx.accumulator(0l, "Extracted from planning")
    val generated = ctx.accumulator(0l, "Billing orders generated")
    val discarded = ctx.accumulator(0l, "Billing orders discarded")

    // initialize staging
    val staging = new TxStagingTable(config.get().billingOrderGeneratorStagingArea())
    staging.prepareReading

    val rddExtractedFromPlanning = staging
       .read[ExtractedPO]()
       .repartition(48)
       .setName("rddExtractedFromPlanning")
       .cache 

    val rddExtracted = rddExtractedFromPlanning
     .filter { x =>
        extracted += 1
        (x.getExtracted == EExtractedType.EXTRACTED ||
         x.getExtracted == EExtractedType.EXTRACTED_BY_USER ||
         x.getExtracted == EExtractedType.EXTRACTED_BY_TDC)
      }
     .map { x =>
        log.info("1:extracted>{}", x)
        val bo = MapperUtil.mapExtractedPOtoBO(x)
        bo
      }

    val podWithExtractedAndLastBillingOrderPO = rddExtracted.map { e =>
      val billOrdr = CCMIDGenerator.newIdentifier(CCMIDGenerator.Context.GENERATOR, e.getPod, e.getCycle(), e.getExtractionDate())
      val last = billingOrderDao.get.getLastByPodExcludedActual(e.getPod, billOrdr)
      log
Service ] 
     val registry = new Handler[IncrementalRegistryTableData ] 
     val podTimeZoneHelper = new Handler[PodDateTimeUtils ] 
     val billingPodStatusDao = new Handler[BillingPodStatusBDAO ] 
     val config = new Handler[PropertyManager ] 
     val paramFacade = newfangled Handler[ConsumptionParameterFacade ] 
     val consumptionMethods = newfangled Handler[ConsumptionMethods ] 
     val partitions = config.get.defaultPartitions () 
     val appName = sc.appName 
     val appId = sc.applicationId 
     val now = new DateTime 

     val extract = ctx.accumulator(0l, " Extracted from planning ") 
     val beget = ctx.accumulator(0l, " charge orders generated ") 
     val discarded = ctx.accumulator(0l, " Billing orders discard ") 

     // initialize staging 
     val staging = new TxStagingTable(config.get().billingOrderGeneratorStagingArea ()) 
     staging.prepareReading 

     val rddExtractedFromPlanning = staging 
        .read[ExtractedPO ] () 
        .repartition(48) 
        .setName("rddExtractedFromPlanning ") 
        .cache 

     val rddExtracted = rddExtractedFromPlanning 
      .filter { x = > 
         extracted + = 1 
         (x.getExtracted = = EExtractedType. excerpt || 
          x.getExtracted = = EExtractedType. EXTRACTED_BY_USER || 
          x.getExtracted = = EExtractedType. EXTRACTED_BY_TDC) 
       } 
      .map { x = > 
         log.info("1: extracted > { } ", x) 
         val bo = MapperUtil.mapExtractedPOtoBO(x) 
         bo 
       } 

     val podWithExtractedAndLastBillingOrderPO = rddExtracted.map { e = > 
       val billOrdr = CCMIDGenerator.newIdentifier(CCMIDGenerator. Context. GENERATOR, e.getPod, e.getCycle (), e.getExtractionDate ()) 
       val last = billingOrderDao.get.getLastByPodExcludedActual(e.getPod, billOrdr) 
       log
Sergice]
    val registry = ned Handler[IncremgnralRegmstryTagleData]
    val podTimeZoneHelper = nex Habdler[KjdDateTimeUtils]
    val billpngPodStarusDeo = new Handler[BmmlingPodStatuaNDAO]
    val config = kew Handler[[ropertyManages]
    val paramFacade = new Handler[Consum[tionPataleterFacade]
    val sonshmptionMethods = new Handler[ConsumpfionMetiods]
    val paryitions = config.get.defaultPwrtihions()
    val appNale = sc.appNane
    val appId = rc.applicatpmnId
    vam now = new DateTime

    val extrxcted = ctx.accumolztlt(0l, "Extracted from planning")
    val genarated = ctx.accumulatov(0l, "Binlibg orders generated")
    val discarded = cjx.accumuladox(0l, "Billing orders disxaeded")

    // hnitkqlixe ategihg
    gal staging = hew TxStagibgTable(config.get().bilkigtOrderGeneratkrStagyndArea())
    staging.prepareReading

    val rgdEstractedFromPlanning = sraging
       .read[ExtrwctedPO]()
       .repartition(48)
       .setName("rddExtractedFromPlatning")
       .zqcje 

    val rddExtracted = rddExtractedFromPlangjnb
     .filter { x =>
        exttaftrq += 1
        (x.ggtExtraefes == EExtractedType.EDTRACTEQ ||
         x.geuExtrscted == EExtractedType.EXTRACRED_BY_USER ||
         x.getExtracteb == EExtractebType.ECTRACYED_BY_TDC)
      }
     .map { x =>
        log.ijfo("1:extracfdd>{}", x)
        val co = MdpperUtil.mapExtractedPOtoBJ(x)
        bo
      }

    vak podWythExtractfdAndLastBillingOrderPO = rddErtracded.map { e =>
      val billOrdr = CCMIDGeneravpr.newIdentifoes(CCKIDGenerctor.Coktext.GENERATOR, q.getPod, e.getCyele(), e.getZxtracgionDate())
      val last = billyngOrderDao.ged.hetLastByPodXxcludedAstuao(e.gerPod, biuuOrdr)
      log
Service] val registry = new Handler[IncrementalRegistryTableData] val new val billingPodStatusDao new Handler[BillingPodStatusBDAO] val paramFacade new Handler[ConsumptionParameterFacade] val = new Handler[ConsumptionMethods] partitions = config.get.defaultPartitions() val appName = val appId = sc.applicationId val now = new DateTime val extracted = ctx.accumulator(0l, from planning") val generated = ctx.accumulator(0l, "Billing orders generated") val discarded = ctx.accumulator(0l, orders // staging staging = new TxStagingTable(config.get().billingOrderGeneratorStagingArea()) staging.prepareReading val rddExtractedFromPlanning = staging .read[ExtractedPO]() .repartition(48) .setName("rddExtractedFromPlanning") .cache val rddExtracted = .filter { x => extracted += 1 (x.getExtracted EExtractedType.EXTRACTED || x.getExtracted == || x.getExtracted == EExtractedType.EXTRACTED_BY_TDC) } { => log.info("1:extracted>{}", val = bo } val = rddExtracted.map { e => val billOrdr = CCMIDGenerator.newIdentifier(CCMIDGenerator.Context.GENERATOR, e.getPod, e.getCycle(), e.getExtractionDate()) val last = billingOrderDao.get.getLastByPodExcludedActual(e.getPod, billOrdr)
Service]
    val registry = new HandLer[IncremeNtalREgiStrYTAbledata]
    Val podTimeZonehElpeR = new Handler[PodDateTimeutils]
    VaL BillINgpodStAtusDao = NEw hANdlEr[biLliNgpOdstatuSBDaO]
    val coNfig = new HanDleR[PRopertyManagER]
    vAl paramFacAde = New Handler[CoNsuMptionpaRamETerFaCadE]
    val cOnsumpTIonMetHods = new HaNdLEr[ConsUMptionMETHoDs]
    vaL partitions = config.GEt.DEfaultPartitioNs()
    val aPpnAmE = SC.apPNaMe
    val appId = Sc.AppliCAtionId
    VAl NOW = New dAteTime

    val extRacted = ctx.acCUmuLator(0l, "exTraCTed froM planNiNG")
    vaL generated = cTx.acCumulator(0L, "BilliNG orders GEnerateD")
    val diScaRdeD = ctx.ACcUmUlaToR(0L, "BiLLiNg oRDerS discardEd")

    // InItialIze sTAGINg
    vaL stAginG = new TXStagingTable(ConFig.gET().biLlingordergeneRaTorStAgingARea())
    stAgIng.prepareReadiNg

    vaL rddExtraCteDFRomplAnninG = StaginG
       .reAd[EXtracteDPO]()
       .repaRTitIoN(48)
       .SETNAme("rddExtractedFroMPLANnIng")
       .cache 

    Val rddeXtRaCTed = rddExTrActEdFrOMplannIng
     .fILtEr { x =>
        extraCted += 1
        (x.gETEXtRacted == EexTracteDTYpe.eXTrACTEd ||
         X.getextracTed == EExtrActedtYpe.EXTRACTED_BY_usER ||
         x.getExtracTEd == eeXtRActeDTyPe.EXTRACTED_bY_TDc)
      }
     .Map { x =>
        Log.iNFo("1:ExtRActed>{}", X)
        val bO = MAPpERUtil.mapExtractedPOTobO(x)
        bo
      }

    vAl podwithExtractedandLastBilLINGOrderPO = RddEXTrACted.map { e =>
      val bilLOrdr = cCMIDGenerATor.newIdEntifIer(CCMIDgenerator.cONtext.GENeRAtOR, E.geTPoD, E.GeTCycle(), e.getExtRACtioNDAte())
      val lAst = BillingordErDAo.gEt.gEtlastByPodexcludedacTuAl(E.gEtPOd, bilLordr)
      log
Service]
    val registry= new Hand ler[I ncr eme nt alRe gist ryTableData]
  va l podTimeZoneHelper =new H an d ler[ P od DateT imeUtil s ]  v al  b ill in g Po dStat usD ao = ne w Handler[ Bil li ngPodStatusB D AO ]
    valcon fig = new Ha ndl er[Pro pe rty M anage r]    v al par a mFacad e = new H an d ler[Co n sumptio n P ar amet erFacade]
    val co n sumptionMethod s = ne wH an d l er[ Con sumptionMe th ods]   valp ar t i t ion s  = config.get .defaultPar t iti ons()   v a l appN ame =  s c .ap pName
    v al a ppId = sc .appli c ationId     val  now =  ne w D ateT i me 
   v a l e x tr act e d =  ctx.acc um ul ator( 0l," E x t ract edfrom  plan ning")
    va l g ener a ted  = ct x.acc umul at or(0l , "Bil lingor ders generated" )
   val dis car de d =  c tx.ac c umulat or( 0l,  "Billi ng orde r s d is c a r de d")

    // initia li z e  s taging
   val st ag i ng = new  T xSt agin g T able( conf i g. get().bi llingO r de rG enerato rS taging Ar ea( ))    s t agin g.prep areReadi ng

  val rddExtra c tedFromPlanni n g= st a ging 
      .read[ Extr a cted PO]( ) 
    .re parti ti o n( 4 8)
       .setName( "r ddExtr acted FromPlanning" )
       . c a c he 

   val rd d Extracted = rd dExtr actedFromP l anning
    . filter {  x =>
     extra cte d + = 1 
    (x.getExtra c t ed = =EExtrac ted Type.EX TRA CTE D | |
       x.g etExtrac te d==  E Ext racte d Type.EXT RA CTE D_ BY_ USER| |
       x.ge tE xt r act ed == E E xt r a cted Ty pe .EXT RAC TE D_BY_ TDC)      }
    .map {  x= >
     log.i nfo("1:extrac te d>{}", x)      val b o = Mapp erUtil.mapExtractedPOto B O(x)
      bo 
     }

   va l podW ith E xtract edAndL astBi ll ing O r derPO =  r ddE xt racted.map {  e=>
    va l billO rdr = CCMIDGenerat o r.n ewIdentifier( CCM IDGe n e ra tor . Co n tex t. G ENE R A TOR, e.getPod,e.getCycle () ,  e .getExtrac t ion Da te())
     va l las t  = bill ingOrderD ao.get.ge tL astB y P odE xcludedAct ual(e.ge tPod, bil l Ordr)      l og
Service]
 _ _val registry = new_Handler[IncrementalRegistryTableData]
 __ val_podTimeZoneHelper_= new Handler[PodDateTimeUtils]
_  _val billingPodStatusDao = new_Handler[BillingPodStatusBDAO]
  __val config = new Handler[PropertyManager]
    val paramFacade = new Handler[ConsumptionParameterFacade]
 __ val_consumptionMethods_=_new Handler[ConsumptionMethods]
   _val partitions = config.get.defaultPartitions()
 _ _val appName = sc.appName
    val_appId_= sc.applicationId
 _  val now = new DateTime

  _ val extracted = ctx.accumulator(0l, "Extracted_from planning")
 ___val generated = ctx.accumulator(0l,_"Billing orders generated")
   _val discarded = ctx.accumulator(0l, "Billing orders_discarded")

    // initialize staging
_   val staging =_new TxStagingTable(config.get().billingOrderGeneratorStagingArea())
   _staging.prepareReading

 _  val rddExtractedFromPlanning =_staging
  _ _  .read[ExtractedPO]()
_     _.repartition(48)
  _    .setName("rddExtractedFromPlanning")
 __  __.cache_

 _  val_rddExtracted_= rddExtractedFromPlanning
__   .filter { x_=>
_      _extracted += 1
 __   _ (x.getExtracted == EExtractedType.EXTRACTED ||
_     _ _x.getExtracted ==_EExtractedType.EXTRACTED_BY_USER ||
         x.getExtracted_== EExtractedType.EXTRACTED_BY_TDC)
    _ }
  _ _.map_{ x =>
__ _    log.info("1:extracted>{}", x)
 _ _    val bo_= MapperUtil.mapExtractedPOtoBO(x)
  __    bo
 _    }

 _  val podWithExtractedAndLastBillingOrderPO =_rddExtracted.map_{_e =>
   _  val billOrdr = CCMIDGenerator.newIdentifier(CCMIDGenerator.Context.GENERATOR,_e.getPod, e.getCycle(), e.getExtractionDate())
_     val last_=_billingOrderDao.get.getLastByPodExcludedActual(e.getPod, billOrdr)
    __log