尝试在 Scala 中并行化嵌套循环

Question

我正在使用嵌套循环和外部 jar 比较 scala/spark 中的 2 个数据帧。

for (nrow <- dfm.rdd.collect) {   
  var mid = nrow.mkString(",").split(",")(0)
  var mfname = nrow.mkString(",").split(",")(1)
  var mlname = nrow.mkString(",").split(",")(2)  
  var mlssn = nrow.mkString(",").split(",")(3)  

  for (drow <- dfn.rdd.collect) {
    var nid = drow.mkString(",").split(",")(0)
    var nfname = drow.mkString(",").split(",")(1)
    var nlname = drow.mkString(",").split(",")(2)  
    var nlssn = drow.mkString(",").split(",")(3)  

    val fNameArray = Array(mfname,nfname)
    val lNameArray = Array (mlname,nlname)
    val ssnArray = Array (mlssn,nlssn)

    val fnamescore = Main.resultSet(fNameArray)
    val lnamescore = Main.resultSet(lNameArray)
    val ssnscore =  Main.resultSet(ssnArray)

    val overallscore = (fnamescore +lnamescore +ssnscore) /3

    if(overallscore >= .95) {
       println("MeditechID:".concat(mid)
         .concat(" MeditechFname:").concat(mfname)
         .concat(" MeditechLname:").concat(mlname)
         .concat(" MeditechSSN:").concat(mlssn)
         .concat(" NextGenID:").concat(nid)
         .concat(" NextGenFname:").concat(nfname)
         .concat(" NextGenLname:").concat(nlname)
         .concat(" NextGenSSN:").concat(nlssn)
         .concat(" FnameScore:").concat(fnamescore.toString)
         .concat(" LNameScore:").concat(lnamescore.toString)
         .concat(" SSNScore:").concat(ssnscore.toString)
         .concat(" OverallScore:").concat(overallscore.toString))
    }
  }
}

我希望做的是为外循环添加一些并行性，这样我就可以创建一个 5 个线程池并从外循环的集合中提取 5 条记录，并将它们与内循环的集合进行比较，而不是连续执行此操作。所以结果是我可以指定线程数，在任何给定时间从外部循环的集合处理中针对内部循环中的集合处理 5 条记录。我该怎么做呢？

Answer 1

让我们从分析您在做什么开始。您将dfm 的数据收集给驱动程序。然后，对于您从 dfn 收集数据的每个元素，对其进行转换并计算每对元素的分数。

这在很多方面都存在问题。首先，即使不考虑并行计算，对dfn 的元素的转换也与dfm 的元素一样多次。此外，您为dfm 的每一行收集dfn 的数据。这是很多网络通信（驱动程序和执行程序之间）。

如果您想使用 spark 并行计算，您需要使用 API（RDD、SQL 或数据集）。您似乎想使用 RDD 来执行笛卡尔积（这是 O(N*M)，所以要小心，这可能需要一段时间）。

让我们首先在笛卡尔积之前转换数据，以避免每个元素多次执行它们。另外，为了清楚起见，让我们定义一个案例类来包含您的数据和一个将您的数据帧转换为该案例类的 RDD 的函数。

case class X(id : String, fname : String, lname : String, lssn : String)
def toRDDofX(df : DataFrame) = {
    df.rdd.map(row => {
        // using pattern matching to convert the array to the case class X
        row.mkString(",").split(",") match {
            case Array(a, b, c, d) => X(a, b, c, d)
        } 
    })
}

然后，我使用filter 仅保留分数超过.95 的元组，但您可以使用map、foreach...，具体取决于您打算做什么。

val rddn = toRDDofX(dfn)
val rddm = toRDDofX(dfm)
rddn.cartesian(rddm).filter{ case (xn, xm) => {
    val fNameArray = Array(xm.fname,xn.fname)
    val lNameArray = Array(xm.lname,xn.lname)
    val ssnArray = Array(xm.lssn,xn.lssn)

    val fnamescore = Main.resultSet(fNameArray)
    val lnamescore = Main.resultSet(lNameArray)
    val ssnscore =  Main.resultSet(ssnArray)

    val overallscore = (fnamescore +lnamescore +ssnscore) /3
    // and then, let's say we filter by score
    overallscore > .95
}} 

Answer 2

这不是迭代 spark 数据帧的正确方法。主要关注的是dfm.rdd.collect。如果数据框任意大，您最终会出现异常。这是因为collect 函数本质上将所有数据都带入了主节点。

另一种方法是使用 rdd 的 foreach 或 map 构造。

dfm.rdd.foreach(x => {
    // your logic
}  

现在您正尝试在此处迭代第二个数据帧。恐怕那是不可能的。优雅的方法是加入 dfm 和 dfn 并迭代生成的数据集以计算您的函数。