使用Scala和Spark将具有巨大价值的Map类型列拆分为多行

如何解决使用Scala和Spark将具有巨大价值的Map类型列拆分为多行

这是问题的延续：Combine value part of Tuple2 which is a map,into single map grouping by the key of Tuple2

我现在可以使用reduceByKey来减少行数。

但是现在，在最后的DataFrame ...

例如

(A,{1->100,2->200,3->100})
(B,{4->300,1->500,9->300,11->900,5->900,6-> 111,7-> 222,8-> 333,12-> 444,13->555,19->666})
(C,{6->100,4->200,7->100,8->200,5->800})

...有些行的地图列包含非常大的地图。例如以上B。

我正在尝试将DF写入Azure Cosmos DB Core SQL。在这里，以上DF的每一行都变成了Cosmos DB的1个文档。问题是，如果行大小大于2MB，则Cosmos DB拒绝该请求。

问题：我想将包含巨大地图列的行拆分为多行（以使它们的大小小于2MB）。键列重复不是问题。

最终结果可以是（如果每次都包含5个以上元素时，我将其划分）：

(A,5->900})
(B,{6-> 111,13->555})
(B,{19->666})
(C,5->800})

您可能会问，在上一个问题中，它已经被拆分了，那么为什么我要合并？原因是在上一个问题中，对于B，如果没有 reduceByKey ，我可能有1000行。但是，最后我只需要上面的20行。 1行本来是理想的选择，但是由于Cosmos的限制，我必须创建多个文档（每个文档小于2MB）。

希望我很清楚。请让我知道是否需要任何澄清。

解决方法

我能够通过编写如下自己的自定义代码来解决此问题：

originalDF.rdd.reduceByKey((a,b) => a ++ b).map(row => {
  val indexedMapEntries: Map[Int,(String,String)] = row._2.zipWithIndex.map(mapWithIndex => (mapWithIndex._2,mapWithIndex._1))
  var min = 0
  var max = Math.min(indexedMapEntries.size - 1,9999)
  var proceed = true

  var rowKeyIdToAttributesMapList: ListBuffer[(String,Map[String,String])] = new ListBuffer[(String,String])]()

  while (proceed) {

    var tempMapToHoldEntries = Map[String,String]()
    var i = min
    while (i <= max) {
      var entry: (String,String) = indexedMapEntries.get(i).get
      tempMapToHoldEntries += entry
      i = i + 1
    }
    rowKeyIdToAttributesMapList += ((row._1,tempMapToHoldEntries))
    min = max + 1
    max = Math.min(indexedMapEntries.size - 1,max + 9999)
    if (min > (indexedMapEntries.size - 1))
      proceed = false

  }

  rowKeyIdToAttributesMapList.toList

}).flatMap(x => x).toDF("rowKeyId","attributes")

在这里，originalDF是我上一个问题中的那个（请检查OP）。 10000是rowKeyId的每个地图的最大大小。如果地图大小超过10000，那么我将创建一个具有相同rowKeyId且循环中剩余属性的新行。