比较 Scala Spark 中的两个数组列

Question

我有一个格式的数据帧，下面给出了。

movieId1 | genreList1              | genreList2
--------------------------------------------------
1        |[Adventure,Comedy]       |[Adventure]
2        |[Animation,Drama,War]    |[War,Drama]
3        |[Adventure,Drama]        |[Drama,War]

并尝试创建另一个标志列，以显示流派列表 2 是否是流派列表 1 的子集。

movieId1 | genreList1              | genreList2        | Flag
---------------------------------------------------------------
1        |[Adventure,Comedy]       | [Adventure]       |1
2        |[Animation,Drama,War]    | [War,Drama]       |1
3        |[Adventure,Drama]        | [Drama,War]       |0

我试过这个：

def intersect_check(a: Array[String], b: Array[String]): Int = {
  if (b.sameElements(a.intersect(b))) { return 1 } 
  else { return 2 }
}
def intersect_check_udf =
  udf((colvalue1: Array[String], colvalue2: Array[String]) => intersect_check(colvalue1, colvalue2))
data = data.withColumn("Flag", intersect_check_udf(col("genreList1"), col("genreList2")))

但这会引发错误

org.apache.spark.SparkException：无法执行用户定义的函数。

PS：上述函数(intersect_check(适用于Array秒。

Answer 1

我们可以定义一个udf来计算两列Array列之间的intersection长度，并检查它是否等于第二列的长度。如果是这样，则第二个数组是第一个数组的子集。

此外，udf的输入必须是类WrappedArray[String]，而不是Array[String]：

import scala.collection.mutable.WrappedArray
import org.apache.spark.sql.functions.col
val same_elements = udf { (a: WrappedArray[String], 
                           b: WrappedArray[String]) => 
  if (a.intersect(b).length == b.length){ 1 }else{ 0 }  
}
df.withColumn("test",same_elements(col("genreList1"),col("genreList2")))
  .show(truncate = false)
+--------+-----------------------+------------+----+
|movieId1|genreList1             |genreList2  |test|
+--------+-----------------------+------------+----+
|1       |[Adventure, Comedy]    |[Adventure] |1   |
|2       |[Animation, Drama, War]|[War, Drama]|1   |
|3       |[Adventure, Drama]     |[Drama, War]|0   |
+--------+-----------------------+------------+----+

数据

val df = List((1,Array("Adventure","Comedy"), Array("Adventure")),
              (2,Array("Animation","Drama","War"), Array("War","Drama")),
              (3,Array("Adventure","Drama"),Array("Drama","War"))).toDF("movieId1","genreList1","genreList2")

Answer 2

这是使用subsetOf转换的解决方案

  val spark =
    SparkSession.builder().master("local").appName("test").getOrCreate()
  import spark.implicits._
  val data = spark.sparkContext.parallelize(
  Seq(
    (1,Array("Adventure","Comedy"),Array("Adventure")),
  (2,Array("Animation","Drama","War"),Array("War","Drama")),
  (3,Array("Adventure","Drama"),Array("Drama","War"))
  )).toDF("movieId1", "genreList1", "genreList2")

  val subsetOf = udf((col1: Seq[String], col2: Seq[String]) => {
    if (col2.toSet.subsetOf(col1.toSet)) 1 else 0
  })
  data.withColumn("flag", subsetOf(data("genreList1"), data("genreList2"))).show()

希望这有帮助！

Answer 3

> Spark 3.0+ ( forall (

forall($"genreList2", x => array_contains($"genreList1", x)).cast("int")

<小时 />

完整示例：

val df = Seq(
     (1, Seq("Adventure", "Comedy"), Seq("Adventure")),
     (2, Seq("Animation", "Drama","War"), Seq("War", "Drama")),
     (3, Seq("Adventure", "Drama"), Seq("Drama", "War"))
     ).toDF("movieId1", "genreList1", "genreList2")
val df2 = df.withColumn("Flag", forall($"genreList2", x => array_contains($"genreList1", x)).cast("int"))
df2.show()
// +--------+--------------------+------------+----+
// |movieId1|          genreList1|  genreList2|Flag|
// +--------+--------------------+------------+----+
// |       1| [Adventure, Comedy]| [Adventure]|   1|
// |       2|[Animation, Drama...|[War, Drama]|   1|
// |       3|  [Adventure, Drama]|[Drama, War]|   0|
// +--------+--------------------+------------+----+

Answer 4

一种解决方案可能是利用 Spark 数组内置函数：如果两者之间的交集等于 genreList2，则genreList2是genreList1的子集。在下面的代码中，添加了sort_array操作，以避免具有不同顺序但元素相同的两个数组之间的不匹配。

val spark = {
    SparkSession
    .builder()
    .master("local")
    .appName("test")
    .getOrCreate()
}
import spark.implicits._
import org.apache.spark.sql._
import org.apache.spark.sql.functions._
val df = Seq(
    (1, Array("Adventure","Comedy"), Array("Adventure")),
    (2, Array("Animation","Drama","War"), Array("War","Drama")),
    (3, Array("Adventure","Drama"), Array("Drama","War"))
).toDF("movieId1", "genreList1", "genreList2")
df
.withColumn("flag",
 sort_array(array_intersect($"genreList1",$"genreList2"))
 .equalTo(
   sort_array($"genreList2")
 )
.cast("integer")
)
.show()

输出为

+--------+--------------------+------------+----+
|movieId1|          genreList1|  genreList2|flag|
+--------+--------------------+------------+----+
|       1| [Adventure, Comedy]| [Adventure]|   1|
|       2|[Animation, Drama...|[War, Drama]|   1|
|       3|  [Adventure, Drama]|[Drama, War]|   0|
+--------+--------------------+------------+----+

Answer 5

这也可以在这里工作，并且不使用udf

 import spark.implicits._
 val data = Seq(
        (1,Array("Adventure","Comedy"),Array("Adventure")),
        (2,Array("Animation","Drama","War"),Array("War","Drama")),
        (3,Array("Adventure","Drama"),Array("Drama","War"))
      ).toDF("movieId1", "genreList1", "genreList2")
 data
     .withColumn("size",size(array_except($"genreList2",$"genreList1")))
     .withColumn("flag",when($"size" === lit(0), 1) otherwise(0))
     .show(false)