IQR.scala

// below Q1 – 1.5 × IQR or above Q3 + 1.5 × IQR

val yieldUplift = deltaDF.select("delta_exploit_relative_yield_uplift").collect().map(_.getDouble(0)).toList.sorted

def getMedian(sorted: List[Double]): Double = {
    val len = sorted.size
    if (len % 2 == 0) {
      (sorted(len / 2 - 1) + sorted(len / 2)) / 2
    } else {
      sorted(len / 2)
    }
}

def IQR(sorted: List[Double], factor: Double = 1.5): (Double, Double) = {
    val len = sorted.size
    val median = getMedian(sorted)
    val (q1, q3) = if (len % 2 == 0) {
      val (low, high) = sorted.span(_ < median)
      (getMedian(low), getMedian(high))
    } else {
      val low = sorted.takeWhile(_ < median)
      val high = sorted.dropWhile(_ <= median)
      (getMedian(low), getMedian(high))
    }
    val range = q3 - q1
    (q1 - 1.5 * range, q3 + 1.5 * range)
}

val (lower, upper) = IQR(yieldUplift)
val upperDomains = deltaDF.select("domain").where($"delta_exploit_relative_yield_uplift" > upper).collect.map(_.getString(0))
val lowerDomains = deltaDF.select("domain").where($"delta_exploit_relative_yield_uplift" < lower).collect.map(_.getString(0))