在Scala中平行一个嵌套环

Question

我正在使用嵌套环和外部罐子比较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 DataFrame上迭代的正确方法。主要问题是dfm.rdd.collect。如果数据框架任意大，则最终会出现异常。这是因为collect功能实质上将所有数据带入主节点。

替代方法是使用RDD的foreach或Map构造。

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

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