如何在火花数据帧中的 ISIN 运算符中传递数据帧

Question

我想将具有一组值的数据帧传递给新查询，但它失败了。

1(在这里我选择特定列，以便我可以在下一个查询中传递ISIN下

scala> val managerIdDf=finalEmployeesDf.filter($"manager_id"!==0).select($"manager_id").distinct
managerIdDf: org.apache.spark.sql.DataFrame = [manager_id: bigint]

2( 我的示例数据：

scala> managerIdDf.show
+----------+                                                                    
|manager_id|
+----------+
|     67832|
|     65646|
|      5646|
|     67858|
|     69062|
|     68319|
|     66928|
+----------+

3(当我执行最终查询时，它失败了：

scala> finalEmployeesDf.filter($"emp_id".isin(managerIdDf)).select("*").show
java.lang.RuntimeException: Unsupported literal type class org.apache.spark.sql.DataFrame [manager_id: bigint]  

我也尝试转换为List和Seq但它只生成错误。如下所示，当我尝试转换为Seq并重新运行查询时，它会抛出错误：

scala> val seqDf=managerIdDf.collect.toSeq
seqDf: Seq[org.apache.spark.sql.Row] = WrappedArray([67832], [65646], [5646], [67858], [69062], [68319], [66928])
scala> finalEmployeesDf.filter($"emp_id".isin(seqDf)).select("*").show
java.lang.RuntimeException: Unsupported literal type class scala.collection.mutable.WrappedArray$ofRef WrappedArray([67832], [65646], [5646], [67858], [69062], [68319], [66928])

我也提到了这篇文章，但徒劳无功。这种类型的查询我正在尝试解决 Spark 数据帧中的子查询。这里有人请？

Answer 1

使用 SPARK SQL 的数据帧和临时视图以及自由格式 SQL 的替代方法 - 不要担心逻辑，这只是约定和初始方法的替代方案 - 这应该同样足够：

val df2 = Seq(
("Peter", "Doe", Seq(("New York", "A000000"), ("Warsaw", null))),
("Bob", "Smith", Seq(("Berlin", null))),
("John", "Jones", Seq(("Paris", null)))
).toDF("firstname", "lastname", "cities")
df2.createOrReplaceTempView("persons")
val res = spark.sql("""select * 
from persons 
where firstname
not in (select firstname
from persons
where lastname <> 'Doe')""")
res.show

或

val list = List("Bob", "Daisy", "Peter")
val res2 = spark.sql("select firstname, lastname from persons")
.filter($"firstname".isin(list:_*))
res2.show

或

val query = s"select * from persons where firstname in (${list.map ( x => "'" + x + "'").mkString(",") })"
val res3 = spark.sql(query)
res3.show

或

df2.filter($"firstname".isin(list: _*)).show

或

val list2 = df2.select($"firstname").rdd.map(r => r(0).asInstanceOf[String]).collect.toList
df2.filter($"firstname".isin(list2: _*)).show 

具体而言：

val seqDf=managerIdDf.rdd.map(r => r(0).asInstanceOf[Long]).collect.toList 2) 
finalEmployeesDf.filter($"emp_id".isin(seqDf: _)).select("").show

Answer 2

是的，您不能在isin中传递数据帧。isin需要一些将筛选的值。

如果你想要一个例子，你可以在这里查看我的答案

根据问题更新，您可以进行以下更改，

.isin(seqDf) 

自

.isin(seqDf: _*)