title: Conditional aggregation Spark DataFrame
body: <p>I would like to understand the best way to do an aggregation in Spark in this scenario:</p>
  
  <pre><code>import sqlContext.implicits._  
  import org.apache.spark.sql.functions._
  case class Person(name:String, acc:Int, logDate:String)
  val dateFormat = "dd/MM/yyyy"
  val filterType = // Could has "MIN" or "MAX" depending on a run parameter
  val filterDate = new Timestamp(System.currentTimeMillis)
  
  val df = sc.parallelize(List(Person("Giorgio",20,"31/12/9999"),
                               Person("Giorgio",30,"12/10/2009")
                               Person("Diego",  10,"12/10/2010"),
                               Person("Diego",  20,"12/10/2010"),
                               Person("Diego",  30,"22/11/2011"), 
                               Person("Giorgio",10,"31/12/9999"),
                               Person("Giorgio",30,"31/12/9999"))).toDF()
  
  val df2 = df.withColumn("logDate",unix_timestamp($"logDate",dateFormat).cast(TimestampType))
  
  val df3 = df.groupBy("name").agg(/*conditional aggregation*/)
  df3.show /*Expected output show  below */
  </code></pre>
  
  <p>Basically I want to group all records by <code>name</code> column and then based on the <code>filterType</code> parameter, I want to filter all valid records for a Person, then after filtering, I want to sum all <code>acc</code> values obtaining a final
  <code>DataFrame</code> with <strong>name</strong> and <strong>totalAcc</strong> columns.</p>
  
  <p>For example: </p>
  
  <ul>
  <li>filterType = <strong>MIN</strong> , I want to take all records with having min(logDate) , could be many of them, so basically in this case I completely ignore filterDate param:</li>
  </ul>
  
  <p><code>Diego,10,12/10/2010
   Diego,20,12/10/2010
   Giorgio,30,12/10/2009</code> </p>
  
  <p>Final result expected from aggregation is: <strong>(Diego, 30),(Giorgio,30)</strong></p>
  
  <ul>
  <li>filterType = <strong>MAX</strong> , I want to take all records with logDate > filterDate, I for a key I don't have any records respecting this condition, I need to take records with min(logDate) as done in MIN scenario, so:</li>
  </ul>
  
  <p><code>Diego,   10, 12/10/2010
   Diego,   20, 12/10/2010
   Giorgio, 20, 31/12/9999
   Giorgio, 10, 31/12/9999
   Giorgio, 30, 31/12/9999</code></p>
  
  <p>Final result expected from aggregation is: <strong>(Diego,30),(Giorgio,60)</strong>
  In this case for Diego I didn't have any records with logDate > logFilter, so I fallback to apply MIN scenario, taking just for Diego all records with min logDate.</p>