Почему преобразование withColumn на искровом фрейме данных не проверяет записи из внешнего списка?

Question

Я использую Spark и Scala для целей обучения.Я столкнулся с ситуацией, когда мне нужно сравнить достоверность записей, присутствующих в одном из столбцов искрового фрейма данных.Вот как я создал один фрейм данных "dataframe1":

import sparkSession.implicits._
val dataframe1 = Seq("AB","BC","CD","DA","AB","BC").toDF("col1")

dataframe1:

+----+
|col1|
+----+
|  AB|
|  BC|
|  CD|
|  DA|
|  AB|
|  BC|
+----+

Срок действия записей зависит от условия, если запись "AB" или "BC».Вот моя первая попытка:

val dataframe2 = dataframe1.withColumn("col2", when('col1.contains("AB") or 'col1.contains("BC"), "valid").otherwise("invalid"))

dataframe2:

+----+-------+
|col1|   col2|
+----+-------+
|  AB|  valid|
|  BC|  valid|
|  CD|invalid|
|  DA|invalid|
|  AB|  valid|
|  BC|  valid|
+----+-------+

Но я не думаю, что это хороший способ сделать это, потому что если мне нужно добавить больше допустимых записей, тогда янеобходимо добавить условия в предложение «когда», что увеличит длину кода и нарушит читабельность кода.

Поэтому я попытался поместить все действительные записи в один список и проверить, присутствует ли строка записи в списке,Если он присутствует, то это действительная запись, иначе нет.Вот фрагмент кода для этого испытания:

val validRecList = Seq("AB", "BC").toList
val dataframe3 = dataframe1.withColumn("col2", if(validRecList.contains('col1.toString())) lit("valid") else lit("invalid"))

Но почему-то он не работает, как ожидалось, в результате этого:

+----+-------+
|col1|   col2|
+----+-------+
|  AB|invalid|
|  BC|invalid|
|  CD|invalid|
|  DA|invalid|
|  AB|invalid|
|  BC|invalid|
+----+-------+

Кто-нибудь может сказать мне, что это за ошибкаЯ тут делаю?И любые другие общие предложения для такого сценария.Спасибо.

Answer 1

dataframe3 код не работает, потому что когда мы увидим документацию о функции «withColumn» в наборе данных https://spark.apache.org/docs/2.2.0/api/scala/index.html#org.apache.spark.sql.Dataset

мы увидим, что withColumn получает «String» и «Column» в качестве типа параметра,

Таким образом, этот код

val dataframe3 = dataframe1.withColumn("col2", if(validRecList.contains('col1.toString())) lit("valid") else lit("invalid"))

даст col2 в качестве имени нового столбца, но даст lit("valid") или lit("invalid") в качестве имени столбца.if(validRecList.contains('col1.toString) lit("valid") else lit("invalid") будет выполнен как скала-код, не выполненный как операция набора данных или операция столбца.

Я имею в виду, что if(validRecList.contains('col1.toString) выполняется scala not spark, потому что «недопустимый» результат получен из validRecList not has 'col1 в списке.Но когда вы определите val validRecList = Seq('col1, "AB", "BC"), validRecList.contains('col1) вернет true

Кроме того, оператор IF не поддерживается для Набор данных и Столбец

Если вы хотите создать условие для функции withColumn, вам нужно выразить выражение типа Column следующим образом:

dataframe3.withColumn("isContainRecList", $"col1".isin(validRecList: _*))

this $"col1".isin(validRecList: _*) является выражением типа Column, поскольку оно будет возвращать Column (на основедокументацию) или вы можете использовать when(the_condition, value_if_true, value_if_false).

Итак, я думаю, что важно понимать типы, которые будет работать с искровым движком с нашими данными, если мы не дадим выражение типа Column, оно будетне ссылаться на данные «col1», но будет ссылаться на «col1» как символ scala .

Также, когда вы хотите использовать IF, возможно, вы могли бы создать пользовательские функции,

import org.apache.spark.sql.functions.udf
def checkValidRecList(needle: String): String = if(validRecList.contains(needle)) "valid" else "invalid"

val checkUdf = udf[String, String](checkValidRecList)

val dataframe3 = dataframe1.withColumn("col2", checkUdf('col1))

результат:

scala> dataframe3.show (false)

+----+-------+
|col1|col2   |
+----+-------+
|AB  |valid  |
|BC  |valid  |
|CD  |invalid|
|DA  |invalid|
|AB  |valid  |
|BC  |valid  |
+----+-------+

Но, я думаю, что мы должны использовать помните, что этот UDF-материал не всегдарекомендуемые.

Answer 2

Попробуйте это:

import spark.implicits._
import org.apache.spark.sql.functions._

val dataframe1 = Seq("AB","BC","CD","DA","AB","BC", "XX").toDF("col1").as[(String)]
val validRecList = List("AB", "BC") 

val dataframe2 = dataframe1.withColumn("col2", when($"col1".isin(validRecList: _*), lit("valid")).otherwise (lit("invalid")))
dataframe2.show(false)

возвращает:

+----+-------+
|col1|col2   |
+----+-------+
|AB  |valid  |
|BC  |valid  |
|CD  |invalid|
|DA  |invalid|
|AB  |valid  |
|BC  |valid  |
|XX  |invalid|
+----+-------+