#47 Add Scala 2.13 support

* #47 Add Scala 2.13 support

* Add a scala 2.13 to build matrix

* Adding `log4j2.properties` to test folder

---------

Co-authored-by: David Benedeki <[email protected]>

Author: Zejnilovic

.github/workflows/build.yml

@@ -29,7 +29,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        scala: [2.11.12, 2.12.12]
+        scala: [2.11.12, 2.12.18, 2.13.11]
     name: Scala ${{matrix.scala}}
     steps:
       - name: Checkout code

.github/workflows/jacoco_check.yml

@@ -27,7 +27,7 @@ jobs:
     strategy:
       matrix:
         include:
-          # The project supports Scala 2.11, 2.12
+          # The project supports Scala 2.11, 2.12, 2.13
           # The CI runs all tests suites for all supported Scala versions at build.yml
           # The codebase for all Scala versions is the same, so the coverage is calculated only once
           # Scala 2.12 is chosen since it is supported by the most wide range of Spark versions and

build.sbt

@@ -21,9 +21,10 @@ ThisBuild / name := "spark-data-standardization"
 ThisBuild / organization := "za.co.absa"
 
 lazy val scala211 = "2.11.12"
-lazy val scala212 = "2.12.12"
+lazy val scala212 = "2.12.18"
+lazy val scala213 = "2.13.11"
 
-ThisBuild / crossScalaVersions := Seq(scala211, scala212)
+ThisBuild / crossScalaVersions := Seq(scala211, scala212, scala213)
 ThisBuild / scalaVersion := scala211
 
 ThisBuild / versionScheme := Some("early-semver")

project/Dependencies.scala

@@ -33,11 +33,11 @@ object Dependencies {
     List(
       "org.apache.spark" %% "spark-core" % sparkVersion % Provided,
       "org.apache.spark" %% "spark-sql" % sparkVersion % Provided,
-      "za.co.absa" %% s"spark-commons-spark$sparkVersionUpToMinor" % "0.5.0" % Provided,
-      "za.co.absa" %% "spark-commons-test" % "0.4.0" % Test,
-      "com.typesafe" % "config" % "1.4.1",
+      "za.co.absa" %% s"spark-commons-spark$sparkVersionUpToMinor" % "0.6.1" % Provided,
+      "za.co.absa" %% "spark-commons-test" % "0.6.1" % Test,
+      "com.typesafe" % "config" % "1.4.2",
       "com.github.mrpowers" %% "spark-fast-tests" % sparkFastTestsVersion(scalaVersion) % Test,
-      "org.scalatest" %% "scalatest" % "3.2.2" % Test
+      "org.scalatest" %% "scalatest" % "3.2.15" % Test
     )
   }
 }

src/main/scala/za/co/absa/standardization/ErrorMessage.scala

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2021 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.standardization
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.types.StructType
+import za.co.absa.standardization.ErrorMessage.Mapping
+
+/**
+ * Case class to represent an error message
+ *
+ * @param errType - Type or source of the error
+ * @param errCode - Internal error code
+ * @param errMsg - Textual description of the error
+ * @param errCol - The name of the column where the error occurred
+ * @param rawValues - Sequence of raw values (which are the potential culprits of the error)
+ * @param mappings - Sequence of Mappings i.e Mapping Table Column -> Equivalent Mapped Dataset column
+ */
+case class ErrorMessage(
+                         errType: String,
+                         errCode: String,
+                         errMsg: String,
+                         errCol: String,
+                         rawValues: Seq[String],
+                         mappings: Seq[Mapping] = Seq()
+                       )
+
+object ErrorMessage {
+  case class Mapping(
+                      mappingTableColumn: String,
+                      mappedDatasetColumn: String
+                    )
+
+  val errorColumnName = "errCol"
+  def errorColSchema(implicit spark: SparkSession): StructType = {
+    import spark.implicits._
+    spark.emptyDataset[ErrorMessage].schema
+  }
+}

src/main/scala/za/co/absa/standardization/SchemaValidator.scala

@@ -18,7 +18,7 @@ package za.co.absa.standardization
 
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.types._
-import za.co.absa.spark.commons.errorhandling.ErrorMessage
+import za.co.absa.standardization.ErrorMessage
 import za.co.absa.standardization.types.{TypeDefaults, TypedStructField}
 import za.co.absa.standardization.validation.field.FieldValidationIssue
 
@@ -116,7 +116,7 @@ object SchemaValidator {
             fields += prefixedField
         }
       }
-      fields
+      fields.toSeq
     }
 
     def flattenArray(field: StructField, arr: ArrayType, structPath: String): Seq[FlatField] = {
@@ -128,7 +128,7 @@ object SchemaValidator {
           val prefixedField = FlatField(structPath, field)
           arrayFields += prefixedField
       }
-      arrayFields
+      arrayFields.toSeq
     }
 
     flattenStruct(schema, "")

src/main/scala/za/co/absa/standardization/Standardization.scala

@@ -16,12 +16,10 @@
 
 package za.co.absa.standardization
 
-import org.apache.hadoop.conf.Configuration
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.{Column, DataFrame, SparkSession}
 import org.slf4j.{Logger, LoggerFactory}
-import za.co.absa.spark.commons.errorhandling.ErrorMessage
 import za.co.absa.spark.commons.implicits.StructTypeImplicits.StructTypeEnhancements
 import za.co.absa.standardization.config.{DefaultStandardizationConfig, StandardizationConfig}
 import za.co.absa.standardization.stages.{SchemaChecker, TypeParser}
@@ -30,7 +28,6 @@ import za.co.absa.standardization.udf.{UDFLibrary, UDFNames}
 
 object Standardization {
   private val logger: Logger = LoggerFactory.getLogger(this.getClass)
-  final val DefaultColumnNameOfCorruptRecord = "_corrupt_record"
 
   final val ColumnNameOfCorruptRecordConf = "spark.sql.columnNameOfCorruptRecord"
 
@@ -39,7 +36,7 @@ object Standardization {
                   standardizationConfig: StandardizationConfig = DefaultStandardizationConfig)
                  (implicit sparkSession: SparkSession): DataFrame = {
     implicit val udfLib: UDFLibrary = new UDFLibrary(standardizationConfig)
-    implicit val hadoopConf: Configuration = sparkSession.sparkContext.hadoopConfiguration
+    udfLib.register(sparkSession)
     implicit val defaults: TypeDefaults = standardizationConfig.typeDefaults
 
     logger.info(s"Step 1: Schema validation")
@@ -75,7 +72,7 @@ object Standardization {
   }
 
   private def standardizeDataset(df: DataFrame, expSchema: StructType, stdConfig: StandardizationConfig)
-                                (implicit spark: SparkSession, udfLib: UDFLibrary, defaults: TypeDefaults): DataFrame  = {
+                                (implicit spark: SparkSession, defaults: TypeDefaults): DataFrame  = {
 
     val rowErrors: List[Column] = gatherRowErrors(df.schema)
     val (stdCols, errorCols, oldErrorColumn) = expSchema.fields.foldLeft(List.empty[Column], rowErrors, None: Option[Column]) {
@@ -97,7 +94,7 @@ object Standardization {
   }
 
   private def cleanTheFinalErrorColumn(dataFrame: DataFrame)
-                                      (implicit spark: SparkSession, udfLib: UDFLibrary): DataFrame = {
+                                      (implicit spark: SparkSession): DataFrame = {
     ArrayTransformations.flattenArrays(dataFrame, ErrorMessage.errorColumnName)
       .withColumn(ErrorMessage.errorColumnName, callUDF(UDFNames.cleanErrCol, col(ErrorMessage.errorColumnName)))
   }

src/main/scala/za/co/absa/standardization/StandardizationErrorMessage.scala

@@ -16,7 +16,7 @@
 
 package za.co.absa.standardization
 
-import za.co.absa.spark.commons.errorhandling.ErrorMessage
+import za.co.absa.standardization.ErrorMessage
 import za.co.absa.standardization.config.{ErrorCodesConfig}
 
 object StandardizationErrorMessage {

src/main/scala/za/co/absa/standardization/ValidationIssue.scala

@@ -16,7 +16,9 @@
 
 package za.co.absa.standardization
 
-sealed abstract class ValidationIssue
+sealed abstract class ValidationIssue {
+  val msg: String
+}
 
 case class ValidationWarning(msg: String) extends ValidationIssue
 

src/main/scala/za/co/absa/standardization/stages/TypeParser.scala

@@ -25,7 +25,7 @@ import org.apache.spark.sql.expressions.UserDefinedFunction
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types._
 import org.slf4j.{Logger, LoggerFactory}
-import za.co.absa.spark.commons.errorhandling.ErrorMessage
+import za.co.absa.standardization.ErrorMessage
 import za.co.absa.spark.commons.implicits.ColumnImplicits.ColumnEnhancements
 import za.co.absa.spark.commons.implicits.StructTypeImplicits.StructTypeEnhancements
 import za.co.absa.spark.commons.utils.SchemaUtils
@@ -39,7 +39,7 @@ import za.co.absa.standardization.time.DateTimePattern
 import za.co.absa.standardization.typeClasses.{DoubleLike, LongLike}
 import za.co.absa.standardization.types.TypedStructField._
 import za.co.absa.standardization.types.{ParseOutput, TypeDefaults, TypedStructField}
-import za.co.absa.standardization.udf.{UDFBuilder, UDFLibrary, UDFNames}
+import za.co.absa.standardization.udf.{UDFBuilder, UDFNames}
 
 import scala.reflect.runtime.universe._
 import scala.util.{Random, Try}
@@ -136,7 +136,7 @@ object TypeParser {
                   origSchema: StructType,
                   stdConfig: StandardizationConfig,
                   failOnInputNotPerSchema: Boolean = true)
-                 (implicit udfLib: UDFLibrary, defaults: TypeDefaults): ParseOutput = {
+                 (implicit defaults: TypeDefaults): ParseOutput = {
     // udfLib implicit is present for error column UDF implementation
     val sourceName = SchemaUtils.appendPath(path, field.sourceName)
     val origField = origSchema.getField(sourceName)
@@ -261,7 +261,7 @@ object TypeParser {
     }
   }
 
-  private abstract class PrimitiveParser[T](implicit defaults: TypeDefaults) extends TypeParser[T] {
+  private abstract class PrimitiveParser[T] extends TypeParser[T] {
     override protected def standardizeAfterCheck(stdConfig: StandardizationConfig)(implicit logger: Logger): ParseOutput = {
       val castedCol: Column = assemblePrimitiveCastLogic
       val castHasError: Column = assemblePrimitiveCastErrorLogic(castedCol)