#40: Make UDFs being registered exactly once per SparkSession (#41)

* upgraded to Spark-commons 0.4.0
* using `OncePerSparkSession` as a predecessor for `UDFLibrary` to prevent multiple registration
* added `doc` to `build.yml` script

Author: benedeki

.github/workflows/build.yml

@@ -40,4 +40,4 @@ jobs:
         with:
           java-version: "[email protected]"
       - name: Build and run tests
-        run: sbt ++${{matrix.scala}} test
+        run: sbt ++${{matrix.scala}} test doc

src/main/scala/za/co/absa/standardization/udf/UDFLibrary.scala

@@ -22,48 +22,52 @@ import org.apache.spark.sql.types._
 import org.apache.spark.sql.{Row, SparkSession}
 import za.co.absa.standardization.config.{ErrorCodesConfig, StandardizationConfig}
 import za.co.absa.standardization.udf.UDFNames._
-import za.co.absa.standardization.{ErrorMessage, Mapping}
+import za.co.absa.standardization.ErrorMessage
+import za.co.absa.spark.commons.OncePerSparkSession
 
 import scala.collection.mutable
 import scala.util.{Failure, Success, Try}
 
-class UDFLibrary(stdConfig: StandardizationConfig)(implicit val spark: SparkSession) extends Serializable {
+class UDFLibrary(stdConfig: StandardizationConfig)(implicit spark: SparkSession) extends OncePerSparkSession with Serializable {
 
   private implicit val errorCodes: ErrorCodesConfig = stdConfig.errorCodes
 
-  spark.udf.register(stdCastErr, { (errCol: String, rawValue: String) =>
-    ErrorMessage.stdCastErr(errCol, rawValue)
-  })
+  override protected def register(implicit spark: SparkSession): Unit = {
 
-  spark.udf.register(stdNullErr, { errCol: String => ErrorMessage.stdNullErr(errCol) })
+    spark.udf.register(stdCastErr, { (errCol: String, rawValue: String) =>
+      ErrorMessage.stdCastErr(errCol, rawValue)
+    })
 
-  spark.udf.register(stdSchemaErr, { errRow: String => ErrorMessage.stdSchemaError(errRow) })
+    spark.udf.register(stdNullErr, { errCol: String => ErrorMessage.stdNullErr(errCol) })
 
-  spark.udf.register(arrayDistinctErrors, // this UDF is registered for _spark-hats_ library sake
-    (arr: mutable.WrappedArray[ErrorMessage]) =>
-      if (arr != null) {
-        arr.distinct.filter((a: AnyRef) => a != null)
-      } else {
-        Seq[ErrorMessage]()
-      }
-  )
+    spark.udf.register(stdSchemaErr, { errRow: String => ErrorMessage.stdSchemaError(errRow) })
 
-  spark.udf.register(cleanErrCol,
-                     UDFLibrary.cleanErrCol,
-                     ArrayType.apply(ErrorMessage.errorColSchema, containsNull = false))
+    spark.udf.register(arrayDistinctErrors, // this UDF is registered for _spark-hats_ library sake
+      (arr: mutable.WrappedArray[ErrorMessage]) =>
+        if (arr != null) {
+          arr.distinct.filter((a: AnyRef) => a != null)
+        } else {
+          Seq[ErrorMessage]()
+        }
+    )
+
+    spark.udf.register(cleanErrCol,
+                       UDFLibrary.cleanErrCol,
+                       ArrayType.apply(ErrorMessage.errorColSchema, containsNull = false))
 
-  spark.udf.register(errorColumnAppend,
-                     UDFLibrary.errorColumnAppend,
-                     ArrayType.apply(ErrorMessage.errorColSchema, containsNull = false))
+    spark.udf.register(errorColumnAppend,
+                       UDFLibrary.errorColumnAppend,
+                       ArrayType.apply(ErrorMessage.errorColSchema, containsNull = false))
 
 
-  spark.udf.register(binaryUnbase64,
-    {stringVal: String => Try {
-      Base64.getDecoder.decode(stringVal)
-    } match {
-      case Success(decoded) => decoded
-      case Failure(_) => null //scalastyle:ignore null
-    }})
+    spark.udf.register(binaryUnbase64,
+      {stringVal: String => Try {
+        Base64.getDecoder.decode(stringVal)
+      } match {
+        case Success(decoded) => decoded
+        case Failure(_) => null //scalastyle:ignore null
+      }})
+  }
 }
 
 object UDFLibrary {