使用spark处理地图结构

Question

我有一个文件，其中包含需要处理的地图结构。我使用了下面的代码。我得到了RDD [ROW] .Data的中间结果如下所示。

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)

数据看起来像这样。

  [("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")]

数据是RDD [ROW]格式。这里我只想要年度标记的总和。如果没有年度标记，我想跳过记录。我想要这样的输出。

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

请帮我。

Answer 1

您可以使用udf函数来达到您的要求，甚至不需要转换为rdd。

我用你给出的样本数据作为形成测试dataframe的方法

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")

这给了我

+--------+-----------+-----------------------------------------------------------------------------------------------------------------+----------+
|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|
+--------+-----------+-----------------------------------------------------------------------------------------------------------------+----------+

正如我所说，一个简单的udf函数应该解决你的要求所以udf函数可以如下

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)

你可以用它如下

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

这将给你所需的dataframe

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

我希望答案是有帮助的