Попытка распараллелить вложенный цикл в Scala

Давайте начнем с анализа того, что вы делаете. Вы собираете данные dfm для драйвера. Затем для каждого элемента вы собираете данные из dfn, преобразуете их и вычисляете оценку для каждой пары элементов.

Это проблематично во многих отношениях. Во-первых, даже без учета параллельных вычислений преобразования элементов dfn выполняются столько раз, сколько dfm элементов. Кроме того, вы собираете данные dfn для каждой строки dfm. Это много сетевых коммуникаций (между драйвером и исполнителями).

Если вы хотите использовать spark для распараллеливания вычислений, вам необходимо использовать API (RDD, SQL или наборы данных). Кажется, вы хотите использовать RDD для выполнения декартова произведения (это O (N * M), поэтому будьте осторожны, это может занять некоторое время).

Начнем с преобразования данных перед декартовым произведением, чтобы не выполнять их более одного раза для каждого элемента. Кроме того, для ясности давайте определим класс case, содержащий ваши данные, и функцию, которая преобразует ваши кадры данных в RDD этого класса case.

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
}} 

Это неправильный способ перебора искрового фрейма данных. Основной проблемой является dfm.rdd.collect. Если кадр данных произвольно велик, вы получите исключение. Это связано с тем, что функция collect по существу переносит все данные в главный узел.

Альтернативным способом может быть использование конструкции foreach или map из rdd.

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

Теперь вы пытаетесь повторить второй фрейм данных здесь. Боюсь, это будет невозможно. Элегантный способ состоит в том, чтобы объединить dfm и dfn и выполнить итерацию по полученному набору данных для вычисления вашей функции.