如何在spark dataframe中的isin操作符中传递dataframe

6bc51xsx 于 2021-06-01 发布在 Hadoop

关注(0)|答案(2)|浏览(571)

我想将具有一组值的dataframe传递给新查询，但失败了。
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])

我也提到这个职位，但徒劳。这种类型的查询我尝试用它来解决sparkDataframe中的子查询。有人在吗？

hadoop apache-spark apache-spark-sql bigdata

来源：https://stackoverflow.com/questions/52613706/how-to-pass-dataframe-in-isin-operator-in-spark-dataframe

2条答案

按热度按时间

o4hqfura1#

使用spark sql的dataframes和tempviews以及自由格式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

赞(0）回复(0）举报 2021-06-02

qoefvg9y2#

是的，您不能传入Dataframe isin . isin 需要一些它将根据其进行筛选的值。
如果你想要一个例子，你可以在这里检查我的答案
根据问题更新，您可以进行以下更改，

.isin(seqDf)

到

.isin(seqDf: _*)

赞(0）回复(0）举报 2021-06-01

我来回答

如何在spark dataframe中的isin操作符中传递dataframe

2条答案

相关问题

热门标签

最新问答