fixed existential type error by converting jobfullparameters in execu…

…tion data to existential, and in runETL with the type provided by the end user
vbounyasit · Feb 20, 2020 · d13d68a · d13d68a
1 parent 773d3c3
commit d13d68a
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 12 deletions.
diff --git a/src/main/scala/com/vbounyasit/bigdata/ETL.scala b/src/main/scala/com/vbounyasit/bigdata/ETL.scala
@@ -19,7 +19,7 @@
 
 package com.vbounyasit.bigdata
 
-import com.vbounyasit.bigdata.ETL.{ExecutionData, OptionalJobParameters}
+import com.vbounyasit.bigdata.ETL.{ExecutionData, JobFullExecutionParameters, OptionalJobParameters}
 import com.vbounyasit.bigdata.args.ArgumentsConfiguration
 import com.vbounyasit.bigdata.args.base.OutputArguments
 import com.vbounyasit.bigdata.config.ConfigurationsLoader
@@ -41,7 +41,7 @@ trait ETL[U, V] {
     * @param args The list of arguments to parse
     * @return An ExecutionData object containing all the required parameters
     */
-  def loadExecutionData(args: Array[String]): ExecutionData[_, _, _, _]
+  protected def loadExecutionData(args: Array[String]): ExecutionData
 
   /**
     * Extracts data from a provided sources configuration
@@ -94,11 +94,12 @@ trait ETL[U, V] {
     *
     * @param executionData The ExecutionData object that will be used
     */
-  def runETL[Config, Argument, ConfigInput, ArgumentInput](executionData: ExecutionData[Config, Argument, ConfigInput, ArgumentInput]): Unit = {
+  def runETL[Config, Argument, ConfigInput, ArgumentInput](executionData: ExecutionData): Unit = {
     implicit val spark: SparkSession = executionData.spark
 
     executionData.jobFullExecutionParameters
-      .foreach(jobParameters => {
+      .foreach(jobParametersExistential => {
+        val jobParameters = jobParametersExistential.asInstanceOf[JobFullExecutionParameters[Config, Argument, ConfigInput, ArgumentInput]]
         //extract
         val sources: Sources = extract(
           jobParameters.outputTable.table,
@@ -140,10 +141,10 @@ object ETL {
 
   type EmptyOptionalParameters = OptionalJobParameters[Nothing, Nothing]
 
-  case class ExecutionData[Config, Argument, ConfigInput, ArgumentInput](configurations: ConfigurationsLoader,
-                                                                         jobFullExecutionParameters: Seq[JobFullExecutionParameters[Config, Argument, ConfigInput, ArgumentInput]],
-                                                                         spark: SparkSession,
-                                                                         environment: String)
+  case class ExecutionData(configurations: ConfigurationsLoader,
+                           jobFullExecutionParameters: Seq[JobFullExecutionParameters[_, _, _, _]],
+                           spark: SparkSession,
+                           environment: String)
 
   case class JobFullExecutionParameters[Config, Argument, ConfigInput, ArgumentInput](jobConf: JobConf,
                                                                                       outputTable: TableMetadata,

diff --git a/src/main/scala/com/vbounyasit/bigdata/SparkApplication.scala b/src/main/scala/com/vbounyasit/bigdata/SparkApplication.scala
@@ -60,7 +60,7 @@ abstract class SparkApplication[U, V] extends SparkSessionProvider with ETL[U, V
     * @param args The list of arguments to parse
     * @return An ExecutionData object containing all the required parameters
     */
-  protected def loadExecutionData(args: Array[String]): ExecutionData[_, _, _, _] = {
+  protected def loadExecutionData(args: Array[String]): ExecutionData = {
 
     case class TableMetadataSeq(tables: Seq[TableMetadata])
 

diff --git a/src/main/scala/com/vbounyasit/bigdata/utils/MonadUtils.scala b/src/main/scala/com/vbounyasit/bigdata/utils/MonadUtils.scala
@@ -33,10 +33,10 @@ object MonadUtils {
   }
 
   def getMapSubList[K, V, Error <: ExceptionHandler](keySubSeq: List[K], dataMap: Map[K, V], errorNotFound: ExceptionWithMessage[Error])(implicit keyConverter: K => String): Either[Error, Map[K, V]] = {
-    def toEither[U](option: Option[(K, U)]): Either[Error, (K, U)] = Either.cond(option.isDefined, option.get, errorNotFound(keyConverter(option.get._1)))
+    def toEither[U](key: K, option: Option[(K, U)]): Either[Error, (K, U)] = Either.cond(option.isDefined, option.get, errorNotFound(keyConverter(key)))
     keySubSeq
-      .map(key => dataMap.get(key).map(key -> _))
-      .map(toEither)
+      .map(key => (key, dataMap.get(key).map(key -> _)))
+      .map(e => toEither(e._1, e._2))
       .sequence.map(_.toMap)
   }
 }