CODEWITHSUNDEEP
apache-sparkapache-spark-sql
Khumar
Khumar

Reputation: 336

processing map structure using spark

I have a file which contains map structure which needs to be processed.I have used below code.I got the intermediate result as RDD[ROW].Data shown below.

val conf=new SparkConf().setAppName("student-example").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlcontext = new org.apache.spark.sql.SQLContext(sc)
    val studentdataframe = sqlcontext.read.parquet("C:\\student_marks.parquet")
    studentdataframe.take(4).foreach(println)

Data looks like this.

  [("Name=aaa","sub=math",Map("weekly" -> Array(25,24,23),"quaterly" -> Array(25,20,19),"annual" -> Array(90,95,97)),"2018-02-03")],
  [("Name=bbb","sub=science",Map("weekly" -> Array(25,24,23),"quaterly" -> Array(25,20,19)),"2018-02-03")],
  [("Name=ccc","sub=math",Map("weekly" -> Array(20,21,18),"quaterly" -> Array(25,16,25)),"2018-02-03")],
  [("Name=ddd","sub=math",Map("weekly" -> Array(25,24,23),"quaterly" -> Array(21,19,15),"annual" -> Array(91,86,64)),"2018-02-03")]

Data is in RDD[ROW] format.Here I want the sum of only annual marks.I want to skip the record if annual marks are not there.I want output like this.

Name=aaa|sub=math|282
Name=ddd|sub=math|241

Please help me.

Upvotes: 0

Views: 271

Answers (1)

Ramesh Maharjan
Ramesh Maharjan

Reputation: 41957

You can achieve your requirement by using a udf function and you don't even need to convert into rdd .

I used your given sample data as a way to form test dataframe as 

val studentdataframe = Seq(
  ("Name=aaa","sub=math",Map("weekly" -> Array(25,24,23),"quaterly" -> Array(25,20,19),"annual" -> Array(90,95,97)),"2018-02-03"),
  ("Name=bbb","sub=science",Map("weekly" -> Array(25,24,23),"quaterly" -> Array(25,20,19)),"2018-02-03"),
  ("Name=ccc","sub=math",Map("weekly" -> Array(20,21,18),"quaterly" -> Array(25,16,25)),"2018-02-03"),
  ("Name=ddd","sub=math",Map("weekly" -> Array(25,24,23),"quaterly" -> Array(21,19,15),"annual" -> Array(91,86,64)),"2018-02-03")
).toDF("name", "sub", "marks", "date")

which gave me 

+--------+-----------+-----------------------------------------------------------------------------------------------------------------+----------+
|name    |sub        |marks                                                                                                            |date      |
+--------+-----------+-----------------------------------------------------------------------------------------------------------------+----------+
|Name=aaa|sub=math   |Map(weekly -> WrappedArray(25, 24, 23), quaterly -> WrappedArray(25, 20, 19), annual -> WrappedArray(90, 95, 97))|2018-02-03|
|Name=bbb|sub=science|Map(weekly -> WrappedArray(25, 24, 23), quaterly -> WrappedArray(25, 20, 19))                                    |2018-02-03|
|Name=ccc|sub=math   |Map(weekly -> WrappedArray(20, 21, 18), quaterly -> WrappedArray(25, 16, 25))                                    |2018-02-03|
|Name=ddd|sub=math   |Map(weekly -> WrappedArray(25, 24, 23), quaterly -> WrappedArray(21, 19, 15), annual -> WrappedArray(91, 86, 64))|2018-02-03|
+--------+-----------+-----------------------------------------------------------------------------------------------------------------+----------+

As I said a simple udf function should solve your requirement so the udf function can be as below

import org.apache.spark.sql.functions._
def sumAnnual = udf((annual: Map[String, collection.mutable.WrappedArray[Int]]) => if (annual.keySet.contains("annual")) annual("annual").sum else 0)

and you can use it as below 

studentdataframe.select(col("name"), col("sub"), sumAnnual(col("marks")).as("sum")).filter(col("sum") =!= 0).show(false)

which will give your required dataframe as 

+--------+--------+---+
|name    |sub     |sum|
+--------+--------+---+
|Name=aaa|sub=math|282|
|Name=ddd|sub=math|241|
+--------+--------+---+

I hope the answer is helpful

Upvotes: 1

Related Questions