sviluppare una PL-SQL su Bigquery partendo dal seguente codice spark: 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()
}
}