fai un analisi logica el seguente script: package com.coop.ingestion.main.batch

import com.accenture.util.DBUtil
import com.coop.spark.
import org.apache.spark.internal.Logging
import org.apache.spark.sql.functions.

class AnagCasseStoricoBatchJob(config: SparkFileBatchJobConfig,flowname:String)

extends SparkApplication with Logging {
val tempTableBucket = config.tempTableBucket

override def sparkConfig: Map[String, String] = config.spark

def start(): Unit = {
withSparkSession { (spark) =>

spark.sparkContext.setLogLevel(“ERROR”)

import spark.implicits._



val prefixPath = s"gs://coop_prod_data/db/dwcoop/stg_anag_casse_storico/“

val prefixLengthPath = prefixPath.length+1
val prefixFile = prefixLengthPath+19


try {

val dfAnagCasseStorico = spark.read.option(“multiline”, “true”)
.json(prefixPath)
.withColumn(s"file_name”, (input_file_name().substr(prefixLengthPath, 39)))
.withColumn(s"data_estrazione", unix_timestamp(concat(input_file_name().substr(prefixFile, 4), lit(“-”)
, input_file_name().substr(prefixFile+4, 2), lit(“-”), input_file_name().substr(prefixFile+6, 2),
lit(" “), input_file_name().substr(prefixFile+9, 2), lit(”:“), input_file_name().substr(prefixFile+11, 2),
lit(”:“), lit(“00”)), “yyyy-MM-dd HH:mm:ss”).minus(2 * 60 * 60))
.createOrReplaceTempView(“stg_anag_casse_storico”)


val sqlAnagCasseStorico =
s”“”
|SELECT
|a.cooperativa
|,cast(cast(a.cod_negozio as int) as string) as cod_negozio
|,a.desc_negozio
|,a.dipartimento
|,a.marca
|,a.modello
|,a.asset_type
|,a.tipologia_cassa
|,a.assistita
|,cast(a.numero_cassa as string) as numero_cassa
|,a.location
|,a.reparto
|,a.pagamento_accettato
|,a.tipo_scansione
|,cast(substr(a.data_acquisto,0,10) as date) as data_acquisto
|,cast(substr(a.data_inizio_garanza,0,10) as date) as data_inizio_garanza
|,cast(a.durata_garanza_mesi as int) as durata_garanza_mesi
|,a.codice_cespite
|,a.utilizzo
|,a.area
|,a.zona
|,a.descrizione_infor
|,a.canale_negozio
|,a.dimensione_negozio
|,a.stato
|,cast(a.versione as int) as versione
|,cast(substr(a.data_inizio_validita,0,10) as date) as data_inizio_validita
|,cast(substr(a.data_fine_validita,0,10) as date) as data_fine_validita
|,cast(a.data_estrazione as timestamp) as data_estrazione
|,a.data_load
|,a.file_name
|FROM(
|SELECT
|cooperativa
|,cod_negozio
|,desc_negozio
|,dipartimento
|,marca
|,modello
|,asset_type
|,tipologia_cassa
|,assistita
|,numero_cassa
|,location
|,reparto
|,pagamento_accettato
|,tipo_scansione
|,data_acquisto
|,data_inizio_garanza
|,durata_garanza_mesi
|,codice_cespite
|,utilizzo
|,area
|,zona
|,descrizione_infor
|,canale_negozio
|,dimensione_negozio
|,stato
|,versione
|,data_inizio_validita
|,data_fine_validita
|,data_estrazione
|,cast(CURRENT_TIMESTAMP as timestamp) as data_load
|,file_name
|FROM
|stg_anag_casse_storico) a
|JOIN
|(SELECT
|max(substring(file_name,1,33)) as file_name
|from stg_anag_casse_storico) b
|ON
|(substring(a.file_name, 1, 33)= b.file_name)
“”“.stripMargin

val dfFinal = spark.sql(sqlAnagCasseStorico)


DBUtil.dynOverwriteHiveTab(tableName = “dwcoop.anag_casse_storico”, df = dfFinal)




println(s”------------------ SUMMARY -----------------------------“)
println(s”#AnagCasseStoricoToWrite [{dfFinal.count()}]")
}
catch {
case e: Exception =>
println(s"------------------ SUMMARY -----------------------------")
println(s"#AnagCasseStoricoToWrite [0]")
}

}
}
}

object AnagCasseStoricoBatchJob {

def main(args: Array[String]): Unit = {
val config = SparkFileBatchJobConfig()

if (args.size != 1) {
throw new IllegalArgumentException("Input is required!")
}

val flowname = args(0)

if (!(flowname.equalsIgnoreCase("anag_casse_storico"))) {
throw new IllegalArgumentException(s"Invalid Flowname{flowname} admitted value anag_casse_storico")
}


val batchJob = new AnagCasseStoricoBatchJob(config,flowname)
batchJob.start()
}
}