获取最大的地图

Question

我有一个RDD这样的地图数组：

Map("id" -> 1, "name" -> "punit") 
Map("id" -> 2, "name" -> "naik", "ph_no" -> 123123) 

现在我的目标是编写此阵图的一个CSV文件，该文件看起来像这个：

id,ph_no,name 
1,,punit 
2,123123,naik 

ID 1没有提供ph_no，这就是为什么它在CSV中为空的原因。所以我想遍历这个RDD并找到最大尺寸的Map，这样我就可以通过提取它的关键字来命名标题中的所有字段。

斯卡拉而言这将是：

val x = Array(Map("id" -> 1, "name" -> "punit"),Map("id" -> 2, "name" -> "naik", "ph_no" -> 123123)).maxBy(_.size) 

这将正确地给我：

res0: scala.collection.immutable.Map[String,Any] = Map(id -> 2, name -> naik, ph_no -> 123123) 

我该怎么办呢？

Answer 1

寻找最大尺寸的Map元素可能不够准确，因为它们都不具备所有数据（从示例中判断）。 您可以通过在地图中对所有不同的键进行联合来获取标题列表。 类似于：

val rddOfMaps:RDD[Map[String,Any]] = sc.parallelize(Seq(Map("a"->1, "b"->2, "d"->3),Map("a"->2, "c"->4, "e" -> 1))) 
val headers = rddOfMaps.flatMap(entry => entry.keySet).distinct.collect 
val csvData = rddOfMaps.map(entry => header.map(column => entry.get(column).getOrElse("")).mkString(",")) 

// 1,2,,3, 
// 2,,4,,1 

Answer 2

您可以使用.max()，指定按地图大小排序。

scala> val rdd = sc.parallelize(Array(Map("id" -> 1, "name" -> "punit"),Map("id" -> 2, "name" -> "naik", "ph_no" -> 123123))) 
rdd: org.apache.spark.rdd.RDD[scala.collection.immutable.Map[String,Any]] = ParallelCollectionRDD[0] at parallelize at <console>:27 

scala> val maxMap = rdd.max()(Ordering.by(_.size)) 
maxMap: scala.collection.immutable.Map[String,Any] = Map(id -> 2, name -> naik, ph_no -> 123123) 

通过，因为你与CSV文件的工作方式，你可以通过使用spark-csv感兴趣。