Разделите текст и найдите общие слова в кадре данных Spark

Question

Я работаю в Scala с Spark, и у меня есть фрейм данных, включающий два столбца с текстом.

Эти столбцы имеют формат «term1, term2, term3, ...», и я хочу создатьтретий столбец с общими терминами двух из них.

Например

Col1 
orange, apple, melon
party, clouds, beach

Col2
apple, apricot, watermelon
black, yellow, white

Результат будет

Col3
1
0

То, что я сделал до сих пор,создать udf, который разделяет текст и получить пересечение двух столбцов.

val common_terms = udf((a: String, b: String) => if (a.isEmpty || b.isEmpty) {
      0
    } else {
      split(a, ",").intersect(split(b, ",")).length
    })

А затем на моем фрейме данных

val results = termsDF.withColumn("col3", common_terms(col("col1"), col("col2"))

Но у меня есть следующая ошибка

Error:(96, 13) type mismatch;
 found   : String
 required: org.apache.spark.sql.Column
      split(a, ",").intersect(split(b, ",")).length

Буду признателен за любую помощь, поскольку я новичок в Scala и просто пытаюсь учиться на онлайн-уроках.

РЕДАКТИРОВАТЬ:

val common_authors = udf((a: String, b: String) => if (a != null || b != null) {
      0
    } else {
      val tempA = a.split( ",")
      val tempB = b.split(",")
      if ( tempA.isEmpty || tempB.isEmpty ) {
        0
      } else {
        tempA.intersect(tempB).length
      }
    })

После редактирования, если я попытаюсь termsDF.show() это работает.Но если я сделаю что-то подобное termsDF.orderBy(desc("col3")), тогда я получу java.lang.NullPointerException

Answer 1

В Spark 2.4 sql вы можете получить те же результаты без UDF.Проверьте это:

scala> val df = Seq(("orange,apple,melon","apple,apricot,watermelon"),("party,clouds,beach","black,yellow,white"), ("orange,apple,melon","apple,orange,watermelon")).toDF("col1","col2")
df: org.apache.spark.sql.DataFrame = [col1: string, col2: string]

scala>

scala> df.createOrReplaceTempView("tasos")

scala> spark.sql(""" select col1,col2, filter(split(col1,','), x -> array_contains(split(col2,','),x) ) a1 from tasos """).show(false)
+------------------+------------------------+---------------+
|col1              |col2                    |a1             |
+------------------+------------------------+---------------+
|orange,apple,melon|apple,apricot,watermelon|[apple]        |
|party,clouds,beach|black,yellow,white      |[]             |
|orange,apple,melon|apple,orange,watermelon |[orange, apple]|
+------------------+------------------------+---------------+

Если вы хотите размер, то

scala> spark.sql(""" select col1,col2, filter(split(col1,','), x -> array_contains(split(col2,','),x) ) a1 from tasos """).withColumn("a1_size",size('a1)).show(false)
+------------------+------------------------+---------------+-------+
|col1              |col2                    |a1             |a1_size|
+------------------+------------------------+---------------+-------+
|orange,apple,melon|apple,apricot,watermelon|[apple]        |1      |
|party,clouds,beach|black,yellow,white      |[]             |0      |
|orange,apple,melon|apple,orange,watermelon |[orange, apple]|2      |
+------------------+------------------------+---------------+-------+


scala>

Answer 2

Попробуйте

val common_terms = udf((a: String, b: String) => if (a.isEmpty || b.isEmpty) {
      0
    } else {
        var tmp1 = a.split(",")
        var tmp2 = b.split(",")
      tmp1.intersect(tmp2).length
    })

val results = termsDF.withColumn("col3", common_terms($"a", $"b")).show

разделить (a, ",") свои функции искрового столбца.Вы используете udf, поэтому вам нужно использовать string.split (), которая является функцией scala

После редактирования: измените нулевую проверку на == не! =