Два DataFrame для каждого цикла

Question

Итерация вложенных циклов * DataFrams foreach создает исключение NullPointerException:

def nestedDataFrame(leftDF: DataFrame, riteDF: DataFrame): Unit = {    
    val leftCols: Array[String] = leftDF.columns
    val riteCols: Array[String] = riteDF.columns

    leftCols.foreach { ltColName =>
        leftDF.select(ltColName).foreach { ltRow =>
            val leftString = ltRow.apply(0).toString();
            // Works ... But Same Kind Of Code Below
            riteCols.foreach { rtColName =>
              riteDF.select(rtColName).foreach { rtRow => //Exception
              val riteString = rtRow.apply(0).toString();
              print(leftString.equals(riteString)
            }
        }
    }

  }

ИСКЛЮЧЕНИЕ:

java.lang.NullPointerException в org.apache.spark.sql.Dataset $ .ofRows (Dataset.scala: 77) в org.apache.spark.sql.Dataset.org $ apache $ spark $ sql $ Набор данных $$ withPlan (Dataset.scala: 3406) в org.apache.spark.sql.Dataset.select (Dataset.scala: 1334) в org.apache.spark.sql.Dataset.select (Dataset.scala: 1352)

Что может быть не так и как это исправить?

Answer 1

leftDF.select(ltColName).foreach { ltRow =>

Приведенная выше строка переносит ваш код в блок foreach как задание для исполнителя. Теперь с riteDF.select(rtColName).foreach { rtRow => вы пытаетесь получить доступ к сеансу Spark в исполнителе, что запрещено. Сеанс Spark доступен только на стороне драйвера. В методе ofRow он пытается получить доступ к sparkSession,

val qe = sparkSession.sessionState.executePlan(logicalPlan)

Вы не можете использовать коллекции наборов данных так же, как обычные коллекции Java / Scala, вы должны использовать их по apis, доступным для выполнения задач, например, вы можете присоединиться к ним, чтобы сопоставить дату.

---

В этом случае вы можете выполнить сравнение несколькими способами. Вы можете присоединиться к 2 наборам данных, например,

var joinedDf = leftDF.select(ltColName).join(riteDF.select(rtColName), $"ltColName" === $"rtColName", "inner")

Затем проанализируйте joinedDf. Вы можете даже intersect() два набора данных.