Как найти общие элементы среди двух столбцов массива?

Question

У меня есть два строковых столбца, разделенных запятыми (sourceAuthors и targetAuthors).

val df = Seq(
  ("Author1,Author2,Author3","Author2,Author3,Author1")
).toDF("source","target")

Я хотел бы добавить еще один столбец nCommonAuthors с количеством общих авторов.

Я пытался сделать это следующим образом:

def myUDF = udf { (s1: String, s2: String) =>
  s1.split(",")
  s2.split(",")
  s1.intersect(s2).length
}
val newDF = myDF.withColumn("nCommonAuthors", myUDF($"source", $"target"))

Я получаю следующую ошибку:

Исключение в потоке "main" java.lang.UnsupportedOperationException:Схема для типа Unit не поддерживается

Есть идеи, почему я получаю эту ошибку?Как найти общие элементы среди двух столбцов?

Answer 1

Основываясь на ответе SCouto, я дам вам полное решение, которое сработало для меня:

  def myUDF: UserDefinedFunction = udf(
(s1: String, s2: String) => {
  val splitted1 = s1.split(",")
  val splitted2 = s2.split(",")
  splitted1.intersect(splitted2).length
})

  val spark = SparkSession.builder().master("local").getOrCreate()

  import spark.implicits._

  val df = Seq(("Author1,Author2,Author3","Author2,Author3,Author1")).toDF("source","target")

  df.show(false)

+-----------------------+-----------------------+
|source                 |target                 |
+-----------------------+-----------------------+
|Author1,Author2,Author3|Author2,Author3,Author1|
+-----------------------+-----------------------+

  val newDF: DataFrame = df.withColumn("nCommonAuthors", myUDF('source,'target))

  newDF.show(false)

+-----------------------+-----------------------+--------------+
|source                 |target                 |nCommonAuthors|
+-----------------------+-----------------------+--------------+
|Author1,Author2,Author3|Author2,Author3,Author1|3             |
+-----------------------+-----------------------+--------------+

Answer 2

Если я не понял вашу проблему, существуют стандартные функции, которые могут вам помочь (поэтому вам не нужно писать UDF), то есть split и array_intersect.

С учетом следующего набора данных:

val df = Seq(("Author1,Author2,Author3","Author2,Author3"))
  .toDF("source","target")
scala> df.show(false)
+-----------------------+---------------+
|source                 |target         |
+-----------------------+---------------+
|Author1,Author2,Author3|Author2,Author3|
+-----------------------+---------------+

Вы можете написать следующий структурированный запрос:

val intersect = array_intersect(split('source, ","), split('target, ","))
val solution = df.select(intersect as "common_elements")
scala> solution.show(false)
+------------------+
|common_elements   |
+------------------+
|[Author2, Author3]|
+------------------+

Answer 3

Эта ошибка означает, что ваш udf возвращает единицу (нет возврата вообще, как void un Java)

Попробуйте это.Вы применяете пересечение по оригинальным s1 и S2, а не по разделенным.

def myUDF = udf((s1: String, s2: String) =>{

  val splitted1 = s1.split(",")


  val splitted2= s2.split(",")


splitted1.intersect(splitted2).length

} )