Merge pull request #57 from X-DataInitiative/CNAM-152-FilteringMainRefactoring

CNAM-152-FilteringMainRefactoring

Author: danielpes

src/main/resources/filtering-default.conf

@@ -0,0 +1,69 @@
+default = {
+  env_name = "default"
+  drug_categories = ["A10"]
+  cancer_definition = "broad"
+  disease_code = "C67"
+  mco_death_code = 9
+  limits = {
+    min_year = 1900
+    max_year = 2020
+    min_month = 1
+    max_month = 12
+    min_gender = 1
+    max_gender = 2
+    min_age = 18
+    max_age = 120
+  }
+  dates = {
+    age_reference = [2006, 12, 31, 23, 59, 59]
+  }
+  paths = {
+    input = {
+      dcir = "/shared/flat_data/joins/DCIR"
+      pmsi_mco = "/shared/flat_data/joins/MCO"
+      pmsi_had = "/shared/flat_data/joins/HAD"
+      pmsi_ssr = "/shared/flat_data/joins/SSR"
+      ir_ben = "/shared/flat_data/IR_BEN_R"
+      ir_imb = "/shared/flat_data/IR_IMB_R"
+      ir_pha = "/shared/value_tables/IR_PHA_R"
+      dosages = "/shared/value_tables/DOSE_PER_MOLECULE.CSV"
+    }
+    output = {
+      root = "/shared/filtered_data"
+      patients = "/shared/filtered_data/patients"
+      flat_events = "/shared/filtered_data/flat_events"
+    }
+  }
+}
+
+# For the CNAM environment, we use the default values
+cnam = ${default}
+cnam.env_name = "cnam"
+
+# Overriding only the dates for cmap environment:
+cmap = ${default}
+cmap.env_name = "cmap"
+cmap.dates = {
+  age_reference = [2010, 12, 31, 23, 59, 59]
+}
+
+# Overriding only the paths for test environment:
+test = ${default}
+test.env_name = "test"
+test.paths = {
+  input = {
+    dcir = "src/test/resources/test-input/DCIR.parquet"
+    pmsi_mco = "src/test/resources/test-input/MCO.parquet"
+    pmsi_had = "src/test/resources/test-input/HAD.parquet"
+    pmsi_ssr = "src/test/resources/test-input/SSR.parquet"
+    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"
+  }
+  output = {
+    root = "target/test/output"
+    patients = "target/test/output/patients"
+    flat_events = "target/test/output/flat_events"
+  }
+}

src/main/resources/filtering.conf

@@ -2,15 +2,18 @@ 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 {
 
   Logger.getRootLogger.setLevel(Level.ERROR)
   Logger.getLogger("org").setLevel(Level.ERROR)
   Logger.getLogger("akka").setLevel(Level.ERROR)
-  Logger.getLogger("fr.polytechnique").setLevel(Level.WARN)
+  Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
+  Logger.getLogger("fr.polytechnique").setLevel(Level.INFO)
 
   Locale.setDefault(Locale.US)
   TimeZone.setDefault(TimeZone.getTimeZone("UTC"))
@@ -26,9 +29,23 @@ 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()
 
+  // Expected args are in format "arg1=value1 arg2=value2 ..."
+  def main(args: Array[String]): Unit = {
+    startContext()
+    val sqlCtx = sqlContext
+    val argsMap = args.map(
+      arg => arg.split("=")(0) -> arg.split("=")(1)
+    ).toMap
+    try {
+      run(sqlCtx, argsMap)
+    }
+    finally stopContext()
+  }
+
   def appName: String
-  def main(args: Array[String]): Unit
+  def run(sqlContext: HiveContext, argsMap: Map[String, String]): Option[Dataset[_]]
 }

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

@@ -4,12 +4,10 @@ import org.apache.spark.sql.expressions.Window
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types.{DateType, IntegerType, StringType, TimestampType}
 import org.apache.spark.sql.{Column, DataFrame, Dataset}
-
 import fr.polytechnique.cmap.cnam.utilities.ColumnUtilities._
 
 object DcirPatientTransformer extends Transformer[Patient] with PatientsTransformer {
 
-
   def estimateBirthDateCol(ts1: Column, ts2: Column, birthYear: Column): Column = {
     unix_timestamp(
       concat(
@@ -34,7 +32,6 @@ object DcirPatientTransformer extends Transformer[Patient] with PatientsTransfor
 
   implicit class PatientTransformer(data: DataFrame) {
 
-
     // The birth year for each patient is found by grouping by patientId and birthYear and then
     //   by taking the most frequent birth year for each patient.
     def findBirthYears: DataFrame = {
@@ -105,5 +102,4 @@ object DcirPatientTransformer extends Transformer[Patient] with PatientsTransfor
     dcir.unpersist()
     result
   }
-
 }

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

@@ -1,14 +1,14 @@
 package fr.polytechnique.cmap.cnam.filtering
 
-import org.apache.spark.sql.{Column, Dataset}
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types.TimestampType
+import org.apache.spark.sql.{Column, Dataset}
 
 /**
   * This trait contains the skeleton of the output events and the target disease code
   */
 trait DiseaseTransformer extends Transformer[Event] {
-  final val DiseaseCode  = "C67"
+  final val DiseaseCode: String = FilteringConfig.diseaseCode
 
   protected val outputColumns: List[Column] = List(
     col("patientID"),

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

@@ -7,7 +7,7 @@ import fr.polytechnique.cmap.cnam.utilities.DrugEventsTransformerHelper
 
 object DrugEventsTransformer extends Transformer[Event] {
 
-  val drugCategories = List("A10") // Only anti-diabetics
+  val drugCategories: List[String] = FilteringConfig.drugCategories
 
   val dcirInputColumns: List[Column] = List(
     col("NUM_ENQ").cast(StringType).as("patientID"),

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

@@ -0,0 +1,110 @@
+package fr.polytechnique.cmap.cnam.filtering
+
+import java.sql.Timestamp
+import scala.collection.JavaConverters._
+import org.apache.spark.SparkContext
+import org.apache.spark.sql.SQLContext
+import com.typesafe.config.{Config, ConfigFactory}
+import fr.polytechnique.cmap.cnam.utilities.functions._
+
+object FilteringConfig {
+
+/* Alternative option using vars instead of SQLContext:
+
+  private var _conf: Config = _
+  private var _path: String = ""
+  private var _env: String = "test"
+  final private val defaultConfig = ConfigFactory.parseResources("filtering-default.conf")
+
+  def path = _path
+  def env = _env
+
+  def setPath(path: String): Unit = { _path = path }
+  def setEnv(env: String): Unit = { _env = env }
+  def init(path: String, env: String ): Unit = {
+    _path = path
+    _env = env
+    init()
+  }
+  def init(): Unit = {
+    _conf = {
+      val defaultConfig = ConfigFactory.parseResources("filtering-default.conf")
+      val config = ConfigFactory.parseFile(new java.io.File(path)).withFallback(defaultConfig).resolve()
+      config.getConfig(env)
+    }
+  }
+
+  def conf = _conf
+*/
+
+  private lazy val conf: Config = {
+    // This is a little hacky. In the future, it may be nice to find a better way.
+    val sqlContext = SQLContext.getOrCreate(SparkContext.getOrCreate())
+    val configPath: String = sqlContext.getConf("conf", "")
+    val environment: String = sqlContext.getConf("env", "test")
+
+    val defaultConfig = ConfigFactory.parseResources("filtering-default.conf").resolve().getConfig(environment)
+    val newConfig = ConfigFactory.parseFile(new java.io.File(configPath))
+
+    newConfig.withFallback(defaultConfig).resolve()
+  }
+
+  case class InputPaths(
+    dcir: String,
+    pmsiMco: String,
+    pmsiHad: String,
+    pmsiSsr: String,
+    irBen: String,
+    irImb: String,
+    irPha: String,
+    dosages: String
+  )
+
+  case class OutputPaths(root: String, patients: String, flatEvents: String)
+
+  case class Limits(
+    minYear: Int,
+    maxYear: Int,
+    minMonth: Int,
+    maxMonth: Int,
+    minGender: Int,
+    maxGender: Int,
+    minAge: Int,
+    maxAge: Int
+  )
+
+  case class Dates(ageReference: Timestamp)
+
+  lazy val drugCategories: List[String] = conf.getStringList("drug_categories").asScala.toList
+  lazy val cancerDefinition: String  = conf.getString("cancer_definition")
+  lazy val diseaseCode: String = conf.getString("disease_code")
+  lazy val mcoDeathCode: Int = conf.getInt("mco_death_code")
+  lazy val inputPaths = InputPaths(
+    dcir = conf.getString("paths.input.dcir"),
+    pmsiMco = conf.getString("paths.input.pmsi_mco"),
+    pmsiHad = conf.getString("paths.input.pmsi_had"),
+    pmsiSsr = conf.getString("paths.input.pmsi_ssr"),
+    irBen = conf.getString("paths.input.ir_ben"),
+    irImb = conf.getString("paths.input.ir_imb"),
+    irPha = conf.getString("paths.input.ir_pha"),
+    dosages = conf.getString("paths.input.dosages")
+  )
+  lazy val outputPaths = OutputPaths(
+    root = conf.getString("paths.output.root"),
+    patients = conf.getString("paths.output.patients"),
+    flatEvents = conf.getString("paths.output.flat_events")
+  )
+  lazy val limits = Limits(
+    minYear = conf.getInt("limits.min_year"),
+    maxYear = conf.getInt("limits.max_year"),
+    minMonth = conf.getInt("limits.min_month"),
+    maxMonth = conf.getInt("limits.max_month"),
+    minGender = conf.getInt("limits.min_gender"),
+    maxGender = conf.getInt("limits.max_gender"),
+    minAge = conf.getInt("limits.min_age"),
+    maxAge = conf.getInt("limits.max_age")
+  )
+  lazy val dates = Dates(
+    ageReference = makeTS(conf.getIntList("dates.age_reference").asScala.toList)
+  )
+}