/**
 * 測試資料：
* 1,001,10,2015-10-29
 * 1,001,8.9,2015-10-28
 * 2,002,5,2015-10-27
 * 2,002,3,2015-10-28
 * 3,003,5.9,2015-11-03
 *
 */
object RemoveMultiRecordTest extends App{
private val sep = "\t"
val sc = ContextInit.getSparkContext();
val filepath = "order/multi_record.txt"
val multiRecord = sc.textFile(filepath).map(_.split(",",-1
 
)).cache()

/**
   * 組裝key_value鍵-值對，groupByKey,對同一Key下的列表求最大值
*/
multiRecord.map(line=>(line(0)+sep+line(1),line(2)+sep+line(3))).groupByKey().map(r=>{
val dateIter = r._2.toIterator
var maxRecord = dateIter.next
var maxDateVal = maxRecord.split(sep)(1)

while(dateIter.hasNext){
val tempRecord = dateIter.next()
val 
 
tempDate = tempRecord.split(sep)(1)
if(DateUtil.isBefore(maxDateVal,tempDate)){
        maxDateVal = tempDate
        maxRecord = tempRecord
      }
    }
    Joiner.on(sep).join(r._1,maxRecord)
  }).foreach(println)

/**
   * 組裝key_value鍵-值對，groupByKey,通過自定義排序，按日期欄位排序，獲取key下第一個記錄
*/
implicit val KeyOrdering = new 
 
Ordering[String] {
override def compare(x: String, y: String): Int = {
val date_x = x.split(sep)(1)
val date_y = y.split(sep)(1)
      date_y.compareTo(date_x)
    }
  }
multiRecord.map(line=>(line(0)+sep+line(1),line(2)+sep+line(3))).groupByKey().map(r=>{
val list = r._2.toArray.sorted(KeyOrdering)
    (r._1,list(0))
  }).foreach(println)

/**
   * 簡潔版
*/
val b = multiRecord.groupBy(a=>(a(0),a(1))).map(r=>{
val t = r._2.toArray.maxBy(m=>m(3))
    t.mkString(sep)
  }).foreach(println)
}