如何解决使用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且循环中剩余属性的新行。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。