algorithm - 使用 scala 和 spark 扫描数据的更好方法

Question

问题

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

我需要从上到下遍历 Sequence 中的这些数据，如果当前记录的类型是 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)

我想要附加到中间的所有 R 型记录 workMap计算到那一点；最后的 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 - R 的列表类型记录附有中间 workMap秒。随着数据的增长，reportList也长大了，我遇到了OutOfMemoryException秒。
2. 我必须合并reportList和 workMap记录在相同的数据结构中，然后返回它们。如果有任何其他优雅的方式，我肯定会考虑改变这种设计。

为了完整起见——我使用的是 spark。函数 calcPerPartition作为参数传递给 RDD 上的 mapPartitions。我需要 workMap s 从每个分区稍后做一些额外的计算。

我知道如果我不必返回 workMap s 来自每个分区，问题就变得简单多了，像这样——

...
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(according to this)，有没有其他方法可以避免存储 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
                  }

              }
            }
        }
    }
  }

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