CNAM-164 Added dosage-based cumulative exposures

CNAM-164 Fixes for running at CNAM

CNAM-164 Added option to reuse flatEvents from a previous run

CNAM-164 Fixed tests

CNAM-164 Updated DosageBased aggregation strategy

Author: danielpes

src/main/resources/config/filtering-default.conf

@@ -2,6 +2,7 @@ default = {
   env_name = "default"
   drug_categories = ["A10"]
   cancer_definition = "broad"
+  # reuse_flat_events_path = /path/to/flat-events/
   disease_code = "C67"
   mco_death_code = 9
   limits = {
@@ -56,6 +57,7 @@ default = {
       window = 1 #Number of months to quantile.
       start_threshold = 6 #Number of months within which more than one purchases have to made
       end_threshold = 4 #Number of months during which no purchases of the particular molecule have to be made
+      dosage_level_intervals = [0,100,200,300,400,500] #list of consumption levels in mg / put only 0 when we want all the weights to 1
     }
   }
 

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

@@ -5,7 +5,6 @@ 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 {
 

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

@@ -2,6 +2,7 @@ package fr.polytechnique.cmap.cnam.filtering
 
 import java.sql.Timestamp
 import scala.collection.JavaConverters._
+import scala.util.Try
 import org.apache.spark.SparkContext
 import org.apache.spark.sql.SQLContext
 import com.typesafe.config.{Config, ConfigFactory}
@@ -92,7 +93,7 @@ object FilteringConfig {
     threshold: Int,
     delay: Int
   )
-
+  lazy val reuseFlatEventsPath: Option[String] = Try(conf.getString("reuse_flat_events_path")).toOption
   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")
@@ -152,6 +153,7 @@ object FilteringConfig {
     filterDelayedPatients = conf.getBoolean("filters.delayed_entries"),
     cumulativeExposureWindow = conf.getInt("exposures.cumulative.window"),
     cumulativeStartThreshold = conf.getInt("exposures.cumulative.start_threshold"),
-    cumulativeEndThreshold = conf.getInt("exposures.cumulative.end_threshold")
+    cumulativeEndThreshold = conf.getInt("exposures.cumulative.end_threshold"),
+    dosageLevelIntervals = conf.getIntList("exposures.cumulative.dosage_level_intervals").asScala.toList.map(x => x.toInt)
   )
 }

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

@@ -4,6 +4,7 @@ import org.apache.spark.sql.Dataset
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.hive.HiveContext
 import fr.polytechnique.cmap.cnam.Main
+import fr.polytechnique.cmap.cnam.filtering.FilteringConfig.{InputPaths, OutputPaths}
 import fr.polytechnique.cmap.cnam.utilities.functions._
 
 object FilteringMain extends Main {
@@ -30,10 +31,20 @@ object FilteringMain extends Main {
     FilteringConfig.init()
   */
 
-    val inputPaths = FilteringConfig.inputPaths
-    val outputPaths = FilteringConfig.outputPaths
-    val cancerDefinition = FilteringConfig.cancerDefinition
-    val upperBoundQuantityIrpha = FilteringConfig.limits.maxQuantityIrpha
+    val reuseFlatEventsPath: Option[String] = FilteringConfig.reuseFlatEventsPath
+
+    if (reuseFlatEventsPath.isDefined) {
+      val flatEventsPath = reuseFlatEventsPath.get
+      logger.info(s"Reusing flatEvents from $flatEventsPath")
+      val flatEvents = sqlContext.read.parquet(flatEventsPath).as[FlatEvent]
+      return Some(flatEvents)
+    }
+
+    val inputPaths: InputPaths = FilteringConfig.inputPaths
+    val outputPaths: OutputPaths = FilteringConfig.outputPaths
+    val cancerDefinition: String = FilteringConfig.cancerDefinition
+    val upperBoundQuantityIrpha: Int = FilteringConfig.limits.maxQuantityIrpha
+
     logger.info(s"Running for the $cancerDefinition cancer definition")
 
     logger.info("(Lazy) Extracting sources...")

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

@@ -105,7 +105,7 @@ object CoxExposuresTransformer extends ExposuresTransformer {
         .withColumn("weight", lit(1.0))
     }
 
-    def withPurchaseBasedCumulativeExposure(cumulativePeriod: Integer): DataFrame = {
+    def withPurchaseBasedCumulativeExposure(cumulativePeriod: Int): DataFrame = {
       val window = Window.partitionBy("patientID", "eventId")
       val windowCumulativeExposure = window.partitionBy("patientID", "eventId", "exposureStart")
 
@@ -114,7 +114,7 @@ object CoxExposuresTransformer extends ExposuresTransformer {
       }
 
       val normalizedExposureDate = udf(
-        (normalizedMonth: Integer) => {
+        (normalizedMonth: Int) => {
           val cal: Calendar = Calendar.getInstance()
           cal.setTime(StudyStart)
           cal.add(Calendar.MONTH, normalizedMonth * cumulativePeriod)

src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/DosageBasedWeightAgg.scala

@@ -1,17 +1,36 @@
 package fr.polytechnique.cmap.cnam.filtering.exposures
 
 import java.sql.Timestamp
-import org.apache.spark.sql.DataFrame
+
+import fr.polytechnique.cmap.cnam.filtering.cox.CoxConfig.CoxExposureDefinition
+import org.apache.spark.sql.{Column, DataFrame}
+import org.apache.spark.sql.expressions.Window
 import org.apache.spark.sql.functions._
 
 class DosageBasedWeightAgg(data: DataFrame) extends WeightAggregatorImpl(data) {
 
+  val minPurchases = 5
+
+  private def aggregateWeightImpl(dosageLevelIntervals: List[Int]): DataFrame = {
+
+    val window = Window.partitionBy("patientID", "eventId")
+    val finalWindow = Window.partitionBy("patientID", "eventId", "weight")
+
+    val getLevel = udf{(Quantity:Double) => dosageLevelIntervals.filter(x => x <= Quantity).size}
+
+    data
+      .withColumn("exposureStart", col("start")) // temporary (todo: "first-only" feature in unlimitedPeriodAdder)
+      .withColumn("weight", (sum("weight").over(window.orderBy("exposureStart"))))
+      .withColumn("weight", getLevel(col("weight")))
+      .withColumn("exposureStart", min("exposureStart").over(finalWindow))
+  }
   def aggregateWeight(
-      studyStart: Option[Timestamp],
-      cumWindow: Option[Int],
-      cumStartThreshold: Option[Int],
-      cumEndThreshold: Option[Int]): DataFrame = {
+      studyStart: Option[Timestamp] = None,
+      cumWindow: Option[Int] = None,
+      cumStartThreshold: Option[Int] = None,
+      cumEndThreshold: Option[Int] = None,
+      dosageLevelIntervals: Option[List[Int]]): DataFrame = {
 
-    data.withColumn("weight", lit("dosage-based cumulative weight"))
+    aggregateWeightImpl(dosageLevelIntervals.get)
   }
 }

src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/ExposuresConfig.scala

@@ -15,7 +15,8 @@ case class ExposuresConfig(
     filterDelayedPatients: Boolean,
     cumulativeExposureWindow: Int,
     cumulativeStartThreshold: Int,
-    cumulativeEndThreshold: Int)
+    cumulativeEndThreshold: Int,
+    dosageLevelIntervals: List[Int])
 
 object ExposuresConfig {
   // todo: Remove filters from ExposuresConfig pipeline

src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/ExposuresTransformer.scala

@@ -37,7 +37,7 @@ class ExposuresTransformer(config: ExposuresConfig)
       .where(col("category") === "molecule")
       .withStartEnd(config.minPurchases,config.startDelay,config.purchasesWindow)
       .where(col("exposureStart") !== col("exposureEnd")) // This also removes rows where exposureStart = null
-      .aggregateWeight(Some(config.studyStart), Some(config.cumulativeExposureWindow), Some(config.cumulativeStartThreshold), Some(config.cumulativeEndThreshold))
+      .aggregateWeight(Some(config.studyStart), Some(config.cumulativeExposureWindow), Some(config.cumulativeStartThreshold), Some(config.cumulativeEndThreshold), Some(config.dosageLevelIntervals))
       .dropDuplicates(Seq("patientID", "eventID", "exposureStart", "exposureEnd", "weight"))
       .select(outputColumns: _*)
       .as[FlatEvent]

src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/NonCumulativeWeightAgg.scala

@@ -12,7 +12,8 @@ class NonCumulativeWeightAgg(data: DataFrame) extends WeightAggregatorImpl(data)
       studyStart: Option[Timestamp],
       cumWindow: Option[Int],
       cumStartThreshold: Option[Int],
-      cumEndThreshold: Option[Int]): DataFrame = this.aggregateWeightImpl
+      cumEndThreshold: Option[Int],
+      dosageLevelIntervals: Option[List[Int]]): DataFrame = this.aggregateWeightImpl
 
-  def aggregateWeight: DataFrame = aggregateWeight(None, None, None, None)
+  def aggregateWeight: DataFrame = aggregateWeight(None, None, None, None, None)
 }

src/main/scala/fr/polytechnique/cmap/cnam/filtering/exposures/PurchaseBasedWeightAgg.scala

@@ -21,7 +21,7 @@ class PurchaseBasedWeightAgg(data: DataFrame) extends WeightAggregatorImpl(data)
     }
 
     val normalizeStart = udf(
-      (normalizedMonth: Integer) => {
+      (normalizedMonth: Int) => {
         val cal: Calendar = Calendar.getInstance()
         cal.setTime(studyStart)
         cal.add(Calendar.MONTH, normalizedMonth * cumWindow)
@@ -41,7 +41,8 @@ class PurchaseBasedWeightAgg(data: DataFrame) extends WeightAggregatorImpl(data)
       studyStart: Option[Timestamp],
       cumWindow: Option[Int],
       cumStartThreshold: Option[Int] = None,
-      cumEndThreshold: Option[Int] = None): DataFrame = {
+      cumEndThreshold: Option[Int] = None,
+      dosageLevelIntervals: Option[List[Int]]= None): DataFrame = {
 
     this.aggregateWeightImpl(studyStart.get, cumWindow.get)
   }