for row in df3.rdd.collect():
            d=row.asDict()
            codice_te=d["ID_TIPO_EVENTO"]
            codice_de=d["ID_DETTAGLIO_EVENTO"]
            descr_te,descr_de=self.F_Anag(jsonI,"TIPO_EVENTO",codice_te,"DETTAGLIO_EVENTO",codice_de)
            descr_te=codice_te if descr_te is None else descr_te
            descr_de=codice_de if descr_de is None else descr_de

val selectCase = udf((tc: String, amt: String) =>
  if (Seq("a", "b").contains(tc)) "Y"
  else if (tc == "a" && amt.toInt <= 0) "N"
  else null
)


dataset1.withColumn("REASON", selectCase(col("tc"), col("amt")))

import spark.implicits._

when($"tc" isin ("a", "b"), "Y")
  .when($"tc" === "a" && $"amt" >= 0, "N")