scala - 使用 Spark Scala 检查一个数据框列中的值是否存在于另一个数据框列中

Question

我有 2 个数据帧 df1 和 df2，

df1 的 Name 列的值如 a、b、c 等df2 的列 ID 的值类似于 a、b

如果 df1 中的 Name 列与 df2 中的 Id 列匹配，那么我们需要将匹配状态设为 0。如果没有匹配，则我们需要匹配状态为 1。我知道我可以使用 collect 将 df2 ID 列放入集合中，然后检查 df1 中的 Name 列是否有匹配的条目。

val df1 = Seq(“Rey”, “John”).toDF(“Name”)
val df2 = Seq(“Rey”).toDF(“Id”)

val collect = df2.select("Id").map(r => r.getString(0)).collect.toList 

有点像，

    val df3 = 
    df1.withColumn("match_sts",when(df1("Name").isin(collect).then(0).else(1)))

Expected output
+ — — + — -+
|Name|match_sts|
+ — — + — -+
| Rey| 0  |
|John| 1  |
+ — — + — -+

但我不想在这里使用收集。是否有任何替代方法可用。

Answer 1

with collect 不是你想要的，但它是 DF col --> list 转换的一个众所周知的问题。如果不是一个巨大的列表，那么你可以这样做 - 这实际上有效，你也可以广播列表:

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

val df1 = Seq("Rey", "John", "Donald", "Trump").toDF("Name")
val df2 = Seq("Rey", "Donald").toDF("Id")

val inlist = df2.select("Id").map(r => r.getString(0)).collect.toList

val df3 = df1.withColumn("match_status", when(df1("Name").isin(inlist: _*),1).otherwise(0))
df3.show(false)

即使在使用文件中的停用词过滤输出的经典示例中，它们也是这样做的:

val stopWords = stopWordsInput.flatMap(x => x.split(" ")).map(_.trim).collect.toSet

如果对 Workers 来说太大，则广播。但是我不确定 10 万是多少!!!

另一种方法是使用 Spark SQL，在使用 EXISTS 时依靠 Catalyst 来优化 SQL:

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

val df1 = Seq("Rey", "John", "Donald", "Trump").toDF("Name")
val df2 = Seq("Rey", "Donald").toDF("Id") // This can be read from file and split etc.

// Optimizer converts to better physical plan for performance in general
df1.createOrReplaceTempView("searchlist") 
df2.createOrReplaceTempView("inlist")    
val df3 = spark.sql("""SELECT Name, 1 
                     FROM searchlist A
                    WHERE EXISTS (select B.Id from inlist B WHERE B.Id = A.Name )
                                   UNION
                   SELECT Name, 0 
                     FROM searchlist A
                    WHERE NOT EXISTS (select B.Id from inlist B WHERE B.Id = A.Name )
                """)
df3.show(false)