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

Question

问题

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

我需要从上到下遍历 Sequence 中的这些数据，如果当前记录是 W 类型，则必须将其与映射合并（我们称之为 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)

我想要附加到中间 workMaps 的所有 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
  }

ReportKeyclass是这样的-

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

我寻求帮助的这种方法有两个问题 -

我必须跟踪 reportList - R 类型记录的列表，并附有中间 workMaps。随着数据的增长，reportList 也在增长，我运行变成了 OutOfMemoryException。
我必须将 reportList 和 workMap 记录组合在同一个数据结构中，然后 return 它们。如果有任何其他优雅的方式，我肯定会考虑改变这种设计。

为了完整起见 - 我正在使用 spark。函数 calcPerPartition 作为参数传递给 RDD 上的 mapPartitions。我需要每个分区的 workMaps 稍后再做一些额外的计算。

我知道如果我不必从每个分区 return 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，但我需要查看数据两次。考虑到输入数据集很大，我也想避免这种情况。

但不幸的是我在后面的步骤中需要 workMaps。

那么，有没有更好的方法来解决上面的问题呢？如果我根本无法解决问题 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
                  }

              }
            }
        }
    }
  }

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

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

Better way to scan data using scala and spark

algorithm

scala

apache-spark

side-effects