Process runs shell successfully, but does not upload data into database

I'm creating a process that basically reads a path from a csv file, then generates a dataframe, and applies a filter to fit for the different fields. This dataframe only needs to write to an Oracle database table, it means does not create a destination hdfs path. The process was created in intellij, it is sbt model, in scala. The process runs and the dataframe loads(append mode) successfully when it is executed directly in the spark shell. However, when the jar is created, and the shell is executed with the bash command and nifi, is executed SUCCEDDED, but it does not load data in the database.I tried to execute it in overwrite mode, but answer was that I did not have privileges. I wrote this, because at the top of the code, in the spark configuration part, it refers to an overwrite mode(i don´t put this spark configuration when is executed directly in spark shell), and I want to know if my code its ok to write in a database, or its something related to a lack of privileges. Next, i share the code of the process:

import com.tchile.bigdata.hdfs.Hdfs
import com.tchile.bigdata.utilities.Utilities
import org.apache.spark.SparkConf
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

import java.sql.DriverManager
import java.util.concurrent.TimeUnit

class Process {
  def process(shell_variable: String): Unit = {

    val startTotal = TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis())
    println("[INFO] Inicio de proceso Logistica Carga Input Simple Data")

    // Productivo (Configuracion de Spark)
    val sparkConf = new SparkConf()
      .setAppName("Logistica_Carga_Input_SimpleData")
      .set("spark.sql.sources.partitionOverwriteMode", "dynamic")
      .set("spark.sql.parquet.output.committer.class", "org.apache.parquet.hadoop.ParquetOutputCommitter")
      .set("spark.sql.sources.commitProtocolClass", "org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol")
    val spark = SparkSession
      .builder()
      .config(sparkConf)
      .enableHiveSupport()
      .getOrCreate()
    val sc = spark.sparkContext

    import spark.implicits._

    val hdfs = new Hdfs
    val utils = new Utilities

    println("[INFO] Obteniendo variables desde shell/NIFI")

    // Valores extraídos desde NIFI
    val daysAgo = shell_variable.split(":")(0).toInt // Cantidad de días de reproceso: inicio
    val daysAhead = shell_variable.split(":")(1).toInt // Cantidad de días de reproceso: límite
    val repartition = shell_variable.split(":")(2).toInt

    // Variables auxiliares para reproceso en HDFS
    var pathToProcess = ""
    var pathToDelete = ""
    var deleteStatus = false

    println("[INFO] Obteniendo variables desde Parametros.conf")

    // Valores extraídos de Parametros.conf > exadata
    val driver_jdbc = sc.getConf.get("spark.exadata.driver_jdbc")
    val url_jdbc = sc.getConf.get("spark.exadata.url_jdbc")
    val user_jdbc = sc.getConf.get("spark.exadata.user_jdbc")
    val pass_jdbc = sc.getConf.get("spark.exadata.pass_jdbc")
    val table_name = sc.getConf.get("spark.exadata.table_name")
    val table_owner = sc.getConf.get("spark.exadata.table_owner")
    val table = sc.getConf.get("spark.exadata.table")

    // Valores extraídos de Parametros.conf > path

    //val pathCsv = sc.getConf.get("spark.path.Csv")

    // Cálculo de la fecha de los días atrás
    val startDate = utils.getCalculatedDate("yyyy-MM-dd", -daysAgo)

    // Cálculo de la fecha límite
    val endDate = utils.getCalculatedDate("yyyy-MM-dd", -daysAhead)

    // Se separan el año, el mes y el día a partir de la fecha_atras
    val startDateYear = startDate.substring(0, 4)
    val startDateMonth = startDate.substring(5, 7)
    val startDateDay = startDate.substring(8, 10)

    // Se separan el año, el mes y el día a partir del sub_day
    val endDateYear = endDate.substring(0, 4)
    val endDateMonth = endDate.substring(5, 7)
    val endDateDay = endDate.substring(8, 10)


    // Información para log
    println("[INFO] Reproceso de: " + daysAgo + " días")
    println("[INFO] Fecha inicio: " + startDate)
    println("[INFO] Fecha límite: " + endDate)

    try {

      // ================= INICIO LÓGICA DE PROCESO =================
      //Se crea df a partir de archivo diario input de acuerdo a la ruta indicada
      val df_csv = spark.read.format("csv").option("header","true").option("sep",";").option("mode","dropmalformed").load("/applications/recup_remozo_equipos/equipos_por_recuperar/output/agendamientos_sin_pet_2")

      val df_final = df_csv.select($"RutSinDV".as("RUT_SIN_DV"),
        $"dv".as("DV"),
        $"Agendado".as("AGENDADO"),
        to_date(col("Dia_Agendado"), "yyyyMMdd").as("DIA_AGENDADO"),
        $"Horario_Agendado".as("HORARIO_AGENDADO"),
        $"Nombre_Agendamiento".as("NOMBRE_AGENDAMIENTO"),
        $"Telefono_Agendamiento".as("TELEFONO_AGENDAMIENTO"),
        $"Email".substr(0,49).as("EMAIL"),
        $"Region_Agendamiento".substr(0,29).as("REGION_AGENDAMIENTO"),
        $"Comuna_Agendamiento".as("COMUNA_AGENDAMIENTO"),
        $"Direccion_Agendamiento".as("DIRECCION_AGENDAMIENTO"),
        $"Numero_Agendamiento".substr(0,5)as("NUMERO_AGENDAMIENTO"),
        $"Depto_Agendamiento".substr(0,9).as("DEPTO_AGENDAMIENTO"),
        to_timestamp(col("fecha_registro")).as("FECHA_REGISTRO"),
        to_timestamp(col("Fecha_Proceso")).as("FECHA_PROCESO")
      )
      // ================== FIN LÓGICA DE PROCESO ==================

      // Limpieza en EXADATA
      println("[INFO] Se inicia la limpieza por reproceso en EXADATA")

      val query_particiones = "(SELECT * FROM (WITH DATA AS (select table_name,partition_name,to_date(trim('''' " +
        "from regexp_substr(extractvalue(dbms_xmlgen.getxmltype('select high_value from all_tab_partitions " +
        "where table_name='''|| table_name|| ''' and table_owner = '''|| table_owner|| ''' and partition_name = '''" +
        "|| partition_name|| ''''),'//text()'),'''.*?''')),'syyyy-mm-dd hh24:mi:ss') high_value_in_date_format " +
        "FROM all_tab_partitions WHERE table_name = '" + table_name + "' AND table_owner = '" + table_owner + "')" +
        "SELECT partition_name FROM DATA WHERE high_value_in_date_format > DATE '" + startDateYear + "-" + startDateMonth + "-" + startDateDay + "' " +
        "AND high_value_in_date_format <= DATE '" + endDateYear + "-" + endDateMonth + "-" + endDateDay + "') A)"

      Class.forName(driver_jdbc)
      val db = DriverManager.getConnection(url_jdbc, user_jdbc, pass_jdbc)
      val st = db.createStatement()

      try {
        val consultaParticiones = spark.read.format("jdbc")
          .option("url", url_jdbc)
          .option("driver", driver_jdbc)
          .option("dbTable", query_particiones)
          .option("user", user_jdbc)
          .option("password", pass_jdbc)
          .load()
          .collect()
        for (partition <- consultaParticiones) {
          st.executeUpdate("call " + table_owner + ".DO_THE_TRUNCATE_PARTITION('" + table + "','" + partition.getString(0) + "')")
        }
      } catch {
        case e: Exception =>
          println("[ERROR TRUNCATE] " + e)
      }

      st.close()
      db.close()

      println("[INFO] Se inicia la inserción en EXADATA")

      df_final.filter($"DIA_AGENDADO" >= "2022-08-01")
        .repartition(repartition).write.mode("append")
        .jdbc(url_jdbc, table, utils.jdbcProperties(driver_jdbc, user_jdbc, pass_jdbc))

      println("[INFO] Inserción en EXADATA completada con éxito")
      println("[INFO] Proceso Logistica Carga Input SimpleData")

      val endTotal = TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis()) - startTotal
      println("[INFO] TIEMPO TOTAL EJECUCIÓN: " + utils.secondsToMinutes(endTotal))
      }
    catch {
      case e: Exception =>
        println("[EXCEPTION] " + e)
        val endTotal = TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis()) - startTotal
        println("[INFO] TIEMPO TOTAL EJECUCIÓN (CON ERROR): " + utils.secondsToMinutes(endTotal))
        throw e
    }
  }
}

Now i want to show the shell file :

#!/bin/bash
args=("$@")
export SPARK_MAJOR_VERSION=2;

in=${args[0]}
process_name2=${args[1]}
dir_hdfs=$(echo $in | awk -F ":" '{print$1}')
process_name=Logistica_Carga_Input_SimpleData

if [  -z "$process_name2" ]; then
        process_name2=$process_name
fi

var_log=$(echo $dir_hdfs"/log/part*")

spark-submit \
--master yarn \
--deploy-mode cluster \
--name   "Logistica_Carga_Input_SimpleData" \
--class  com.tchile.bigdata.App \
--num-executors 2 \
--properties-file  "/home/TCHILE.LOCAL/srv_remozo_equip/ingesta/Logistica_Carga_Input_SimpleData/Parametros.conf" \
--conf   "spark.executor.cores=16" \
--conf   "spark.default.parallelism=120" \
--conf   "spark.driver.memory=20gb" \
--conf   "spark.executor.memory=30gb" \
--conf   "spark.ui.acls.enable=true" \
--conf   "spark.sql.sources.partitionOverwriteMode=dynamic" \
--conf   "spark.ui.view.acls=srv_remozo_equip" \
--conf   "spark.admin.acls=srv_remozo_equip" \
--conf   "spark.sql.warehouse.dir=/etc/hive/conf/hive-site.xml" \
--conf   "spark.executor.extraJavaOptions=-Djava.security.auth.login.config=./kafka_client_jaas.conf" \
--conf   "spark.driver.extraJavaOptions=-Djava.security.auth.login.config=./kafka_client_jaas.conf" \
--conf   "spark.driver.extraJavaOptions=-Dlog4j.configuration=./log4j.properties" \
--conf   "spark.executor.extraJavaOptions=-Dlog4j.configuration=./log4j.properties" \
--conf "spark.driver.extraClassPath=/usr/share/java/ojdbc7.jar" \
--jars "/usr/share/java/ojdbc7.jar" \
hdfs://nn:8020/applications/recup_remozo_equipos/Logistica_Carga_Input_SimpleData/logistica_carga_input_simpledata_2.11-1.0.jar "${args[0]}"

Some comments, field names are in Spanish, but I don't think it's relevant for the analysis. I also do not include the variables configuration parameter file. I appreciate any comments on this.