CNAM-152 Refactored run signature to return an optional Dataset

CNAM-152 Corrected csv file extension

CNAM-152 Small changes and review

Author: danielpes

src/main/resources/filtering-default.conf

@@ -59,7 +59,7 @@ test.paths = {
     ir_ben = "src/test/resources/test-input/IR_BEN_R.parquet"
     ir_imb = "src/test/resources/test-input/IR_IMB_R.parquet"
     ir_pha = "src/test/resources/test-input/IR_PHA_R.parquet"
-    dosages = "src/test/resources/test-input/DOSE_PER_MOLECULE.csv"
+    dosages = "src/test/resources/test-input/DOSE_PER_MOLECULE.CSV"
   }
   output = {
     root = "target/test/output"

src/main/scala/fr/polytechnique/cmap/cnam/Main.scala

@@ -2,8 +2,10 @@ package fr.polytechnique.cmap.cnam
 
 import java.util.{Locale, TimeZone}
 import org.apache.log4j.{Level, Logger}
+import org.apache.spark.sql.Dataset
 import org.apache.spark.sql.hive.HiveContext
 import org.apache.spark.{SparkConf, SparkContext}
+import fr.polytechnique.cmap.cnam.flattening.FlatteningMain._
 
 trait Main {
 
@@ -27,6 +29,7 @@ trait Main {
   def startContext(): Unit = {
     _sc = new SparkContext(new SparkConf().setAppName(this.appName))
     _sql = new HiveContext(_sc)
+    _sql.setConf("spark.sql.autoBroadcastJoinThreshold", "104857600")
   }
   def stopContext(): Unit = _sc.stop()
 
@@ -44,5 +47,5 @@ trait Main {
   }
 
   def appName: String
-  def run(sqlContext: HiveContext, argsMap: Map[String, String]): Unit = {}
+  def run(sqlContext: HiveContext, argsMap: Map[String, String]): Option[Dataset[_]]
 }

src/main/scala/fr/polytechnique/cmap/cnam/filtering/FilteringMain.scala

@@ -15,7 +15,7 @@ object FilteringMain extends Main {
     *   "conf" -> "path/to/file.conf" (default: "$resources/filtering-default.conf")
     *   "env" -> "cnam" | "cmap" | "test" (deafult: "test")
     */
-  override def run(sqlContext: HiveContext, argsMap: Map[String, String] = Map()): Unit = {
+  def run(sqlContext: HiveContext, argsMap: Map[String, String] = Map()): Option[Dataset[FlatEvent]] = {
 
     import implicits.SourceExtractor
     import sqlContext.implicits._
@@ -70,5 +70,7 @@ object FilteringMain extends Main {
     patients.toDF.write.parquet(outputPaths.patients)
     logger.info("Writing FlatEvents...")
     flatEvents.toDF.write.parquet(outputPaths.flatEvents)
+
+    Some(flatEvents)
   }
 }

src/main/scala/fr/polytechnique/cmap/cnam/filtering/PatientsTransformer.scala

@@ -20,7 +20,7 @@ trait PatientsTransformer {
 object PatientsTransformer extends Transformer[Patient] with PatientsTransformer {
 
   def isDeathDateValid(deathDate: Column, birthDate: Column): Column =
-    deathDate.between(birthDate, lit(makeTS(MaxYear, 1, 1)))
+    deathDate.between(birthDate, makeTS(MaxYear, 1, 1))
 
   def transform(sources: Sources): Dataset[Patient] = {
     val irBen = IrBenPatientTransformer.transform(sources).toDF.as("irBen")

src/main/scala/fr/polytechnique/cmap/cnam/filtering/cox/CoxMain.scala

@@ -121,4 +121,7 @@ object CoxMain extends Main {
     coxFeaturing(sqlContext, config, cancerDefinition, filterDelayedPatients)
     stopContext()
   }
+
+  // todo: refactor this function
+  def run(sqlContext: HiveContext, argsMap: Map[String, String]): Option[Dataset[_]] = None
 }

src/main/scala/fr/polytechnique/cmap/cnam/filtering/ltsccs/LTSCCSMain.scala

@@ -1,5 +1,6 @@
 package fr.polytechnique.cmap.cnam.filtering.ltsccs
 
+import org.apache.spark.sql.Dataset
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.hive.HiveContext
 import com.typesafe.config.{Config, ConfigFactory}
@@ -60,4 +61,7 @@ object LTSCCSMain extends Main {
     runLTSCCS(sqlContext, config)
     stopContext()
   }
+
+  // todo: refactor this function
+  def run(sqlContext: HiveContext, argsMap: Map[String, String]): Option[Dataset[_]] = None
 }

src/main/scala/fr/polytechnique/cmap/cnam/filtering/mlpp/MLPPMain.scala

@@ -1,6 +1,7 @@
 package fr.polytechnique.cmap.cnam.filtering.mlpp
 
 import scala.collection.JavaConversions._
+import org.apache.spark.sql.Dataset
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.hive.HiveContext
 import com.typesafe.config.{Config, ConfigFactory}
@@ -49,4 +50,7 @@ object MLPPMain extends Main {
     MLPPFeaturing(sqlContext, config)
     stopContext()
   }
+
+  // todo: refactor this function
+  def run(sqlContext: HiveContext, argsMap: Map[String, String]): Option[Dataset[_]] = None
 }

src/main/scala/fr/polytechnique/cmap/cnam/flattening/FlatteningMain.scala

@@ -1,6 +1,7 @@
 package fr.polytechnique.cmap.cnam.flattening
 
-import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.{DataFrame, Dataset}
+import org.apache.spark.sql.hive.HiveContext
 import fr.polytechnique.cmap.cnam.Main
 import fr.polytechnique.cmap.cnam.utilities.FlatteningConfig
 import fr.polytechnique.cmap.cnam.utilities.FlatteningConfig._
@@ -60,10 +61,9 @@ object FlatteningMain extends Main {
     }
   }
 
-  override def main(args: Array[String]): Unit = {
-    startContext()
-    sqlContext.setConf("spark.sql.autoBroadcastJoinThreshold", "104857600")
+  def run(sqlContext: HiveContext, argsMap: Map[String, String]): Option[Dataset[_]] = {
     loadToParquet()
     joinTables()
+    None
   }
 }

src/main/scala/fr/polytechnique/cmap/cnam/flattening/ValidateFlattening.scala

@@ -3,8 +3,9 @@ package fr.polytechnique.cmap.cnam.flattening
 import fr.polytechnique.cmap.cnam.Main
 import fr.polytechnique.cmap.cnam.statistics.Comparator
 import fr.polytechnique.cmap.cnam.utilities.RichDataFrames._
-import org.apache.spark.sql.{Column, DataFrame}
+import org.apache.spark.sql.{Column, DataFrame, Dataset}
 import org.apache.spark.sql.functions._
+import org.apache.spark.sql.hive.HiveContext
 import fr.polytechnique.cmap.cnam.utilities.FlatteningConfig
 import fr.polytechnique.cmap.cnam.utilities.FlatteningConfig._
 
@@ -116,8 +117,8 @@ object ValidateFlattening extends Main {
     }
   }
 
-  override def main(args: Array[String]){
-    startContext( )
+  def run(sqlContext: HiveContext, argsMap: Map[String, String]): Option[Dataset[_]] = {
     computeStoreFlatAndInputDfsStat()
+    None
   }
 }