使用 scala 和 Spark 扫描数据的更好方法

Question

>问题

输入数据有 2 种类型的记录，我们称它们为R和W。

我需要按顺序从上到下遍历这些数据，如果当前记录的类型为W，则必须将其与 map 合并(我们称之为workMap)。如果映射中已存在该 W 类型记录的键，则会将此记录的值添加到其中，否则会在workMap中创建新条目。

如果当前记录的类型为R，则在此记录之前计算workMap将附加到当前记录。

例如，如果这是记录的顺序 -

W1-   a -> 2
W2-   b -> 3
W3-   a -> 4
R1 
W4-   c -> 1
R2
W5-   c -> 4

其中 W1、W2、W3、W4 和 W5 属于W型;R1 和 R2 属于R型

在这个函数的最后，我应该有以下内容——

R1 - { a -> 6, 
b -> 3 } //merged(W1, W2, W3)
R2 - { a -> 6, 
b -> 3,
c -> 1 } //merged(W1, W2, W3, W4)
{ a -> 6, 
b -> 3,
c -> 5 } //merged(W1, W2, W3, W4, W5)

我希望所有附加到中间workMap的 R 类型记录都计算到那时;以及处理最后一条记录后的最终workMap。

这是我写的代码——

def calcPerPartition(itr: Iterator[(InputKey, InputVal)]):
Iterator[(ReportKey, ReportVal)] = {
val workMap = mutable.HashMap.empty[WorkKey, WorkVal]
val reportList = mutable.ArrayBuffer.empty[(ReportKey, Reportval)]
while (itr.hasNext) {
val temp = itr.next()
val (iKey, iVal) = (temp._1, temp._2)
if (iKey.recordType == reportType) {
//creates a new (ReportKey, Reportval)
reportList += getNewReportRecord(workMap, iKey, iVal) 
}
else {
//if iKey is already present, merge the values 
//other wise adds a new entry
updateWorkMap(workMap, iKey, iVal) 
}
}
val workList: Seq[(ReportKey, ReportVal)] = workMap.toList.map(convertToReport)
reportList.iterator ++ workList.iterator
}

ReportKey课是这样的——

case class ReportKey (
// the type of record - report or work 
rType: Int, 
date: String, 
.....
)

---

我正在寻求帮助的这种方法有两个问题——

1. 我必须跟踪一个reportList- 附有中间workMap的R类型记录的列表。随着数据的增长，reportList也在增长，我遇到了OutOfMemoryException。
2. 我必须将reportList和workMap记录组合在同一数据结构中，然后返回它们。如果还有其他优雅的方式，我肯定会考虑改变这个设计。

为了完整起见 - 我正在使用火花。函数calcPerPartition作为 RDD 上 mapPartition 的参数传递。我需要每个分区的workMap，以便稍后进行一些额外的计算。

我知道如果我不必从每个分区返回workMaps，问题就会变得容易得多，就像这样 -

...
val workMap = mutable.HashMap.empty[WorkKey, WorkVal]                     
itr.scanLeft[Option[(ReportKey, Reportval)]](
None)((acc: Option[(ReportKey, Reportval)], 
curr: (InputKey, InputVal)) => {
if (curr._1.recordType == reportType) {
val rec = getNewReportRecord(workMap, curr._1, curr._2)
Some(rec)
}
else {
updateWorkMap(workMap, curr._1, curr._2)
None
}
})
val reportList = scan.filter(_.isDefined).map(_.get)
//workMap is still empty after the scanLeft. 
... 

当然，我可以对输入数据执行reduce操作以得出最终workMap但我需要查看数据两次。考虑到输入数据集很大，我也想避免这种情况。

但不幸的是，我需要后一步的workMap。

那么，有没有更好的方法来解决上述问题呢？如果我根本无法解决问题2(根据这一点)，是否有其他方法可以避免将R记录(reportList)存储在列表中或多次扫描数据？

Answer 1

对于第二个问题，我还没有更好的设计 - 如果您可以避免将reportList和workMap组合成单个数据结构，但我们当然可以避免将R类型记录存储在列表中。

以下是我们如何重写上述问题中的calcPerPartition-

def calcPerPartition(itr: Iterator[(InputKey, InputVal)]):
Iterator[Option[(ReportKey, ReportVal)]] = {
val workMap = mutable.HashMap.empty[WorkKey, WorkVal]
var finalWorkMap = true
new Iterator[Option[(ReportKey, ReportVal)]](){
override def hasNext: Boolean = itr.hasNext
override def next(): Option[(ReportKey, ReportVal)] = {
val curr = itr.next()
val iKey = curr._1
val iVal = curr._2
val eventKey = EventKey(openKey.date, openKey.symbol)
if (iKey.recordType == reportType) {
Some(getNewReportRecord(workMap, iKey, iVal))
}
else {
//otherwise update the generic interest map but don't accumulate anything
updateWorkMap(workMap, iKey, iVal)
if (itr.hasNext) {
next()
}
else {
if(finalWorkMap){
finalWorkMap = false //because we want a final only once
Some(workMap.map(convertToReport))
}
else {
None
}
}
}
}
}
}

我们没有将结果存储在列表中，而是定义了一个迭代器。这解决了我们围绕此问题遇到的大部分内存问题。