import org.apache.spark._
import org.apache.spark.SparkContext._
import org.apache.log4j._
import sun.applet.Main

object Ex_4_4 {

    def getDatailsOfAge(linha: String)=
    {
        val brokenDbAge=linha.split(",")
        ((brokenDbAge(1),brokenDbAge(3),brokenDbAge(5)),1)
    }

    def main (args: Array[String])
    {
        Logger.getLogger("org").setLevel(Level.ERROR)

        // Load and map children
        val sc                    = new SparkContext("local[*]","banco")
        val loadBase              = sc.textFile("/Users/saulostopa/Scala/Puc-Minas-PPDD/ProjetosSparkScala/projeto1/bank-data.csv")
        val mapedBaseDatailsOfAge = loadBase.map(getDatailsOfAge)

        val result                = mapedBaseDatailsOfAge.reduceByKey((x,y)=>x+y).filter(x=>Integer.parseInt(x._1._1)<=30).sortBy(_._1,true,1).sortBy(_._2,true,1)

        result.foreach(println)
        //print(result)
    }
}