I have the following data: 

```
53,Male,11th,<=50K 
53,Male,11th,<=50K  
53,Male,11th,<=50K
20,Female,Masters,>50K 
20,Female,Masters,>50K
33,Male,Bachelors,<=50K

```

Next I need to group the above data by using select and group. so it will be something like: 

```
53,Male,11th,<=50K,3 
20,Female,Masters,>50K,2 
33,Male,Bachelors,<=50K,1

```

where the last number shows the number of similar records. Now I need to filter the number of equivalent records > 2, and store it in a separate file

I have grouped data by sql queries in Scala command. To ungroup data, I though we may create a table and add the grouped data by (insert command), and line by line. It works, but this is very very slow, and took around one hour for few records. Is there any thoughts using Scala with many thanks.
The command is shown below:

```
import spark.sqlContext.implicits._
import scala.collection.immutable.Map
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType}

case class Rating(age: Double,edu: String, sex: String, salary: String)

val Result = sc.textFile("hdfs://NameNode01:9000/input/adult.csv").map(_.split(",")).map(p => Rating(p(0).trim.toDouble,p(1),p(2),p(3))).toDF()
Result.registerTempTable("Start")

val sal1=spark.sqlContext.sql("SELECT  age,edu,sex,salary,count(*) as cnt from Start group by age,edu,sex,salary")
sal1.registerTempTable("adult")

val sal2=spark.sqlContext.sql("SELECT  age,edu,sex,salary,cnt from adult WHERE cnt>3")
sal2.registerTempTable("adult2")

var ag=sal2.map(age => ""+age(0)).collect()
var ed=sal2.map(edu => ""+edu(1)).collect()
var se=sal2.map(sex => ""+sex(2)).collect()
var sa=sal2.map(salary => ""+salary(3)).collect()
var cn=sal2.map(cnt => ""+cnt(4)).collect()

//convert age to double
val ages= ag.map(_.toDouble)

//convert the cnt to integer
val counts= cn.map(_.toInt)

//length of the array
var cnt_length=counts.size

//create a table and add the sal2 records in it
val adlt2=spark.sqlContext.sql("CREATE TABLE adult3 (age double, edu string, sex string, salary string)")

//loop and enter the number of cn
var sql_querys="query"
var i=0
var j=0
var loop_cnt=0

for(i <-0 to cnt_length-1){
   loop_cnt=counts(i)
    for(j <-0 to loop_cnt-1){
        sql_querys="INSERT into adult3 values ("+ages(i)+",'"+ed(i)+"','"+se(i)+"','"+sa(i)+"')"

        val adlt3=spark.sqlContext.sql("INSERT into adult3 values ("+ages(i)+",'"+ed(i)+"','"+se(i)+"','"+sa(i)+"')")
    }

}

```

The main part is the loop at the end of the code.