not able to merge spark scala code in spark pipeline

we are merging our spark scala code with spark pipeline, however if i run the code step individually it runs fine (in both function mode / free mode) but when it is been merged in spark pipeline it give error

java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Driver stacktrace:, caused by: ClassNotFoundException: CustomScalaRecipe$$anonfun$1

scala code:

import org.apache.spark.sql.functions._

// Recipe inputs
val iups_prepared_joined_by_time_cell_joined = inputDatasets("iups_prepared_joined_by_time_cell_joined")


def redaction(data:String, d_cnt: Int): String = {

     var data_trim = "" ;
     if (data.isEmpty || data =="" || data == null){
         return data ;
     }else {   
         data_trim = data.trim()  ;
     }

     if (d_cnt == 1){
         var data_length = data_trim.length();
         return data_trim.replaceAll(data_trim, "*"*data_length)
     }
     else if (d_cnt> 5){
         return data_trim
     }
     else {
         val redac_length = 6-d_cnt
         val length_r = Seq(redac_length, data_trim.length()).min
         return data_trim.slice(0,data_trim.length()-length_r) + "*"*(length_r+3)
    }

 }

val redaction_udf =  udf((data: String,d_cnt:Int) => redaction(data,d_cnt));

val iups_pre_anonymized = iups_prepared_joined_by_time_cell_joined.withColumn("price_plan", when(col("price_plan").isNull, lit(null)).otherwise(redaction_udf(col("price_plan"),col("price_plan_distinct"))))
// Recipe outputs

Map("iups_pre_anonymized" -> iups_pre_anonymized)