比较Spark 2的DataFrame
比较Spark 2的DataFrame
提问于 2018-01-25 22:58:38
我是斯派克/斯卡拉的新手。我有一个由1亿多条记录组成的主数据框架。
+--------+
| ttm_id|
+--------+
|39622109|
|39622178|
|39578322|
+--------+还有一个改变派DataFrame，它有大约4000万张唱片
+----------+--------+
|__change__| ttm_id|
+----------+--------+
| DELETE|18001570|
| DELETE| 50520|
| DELETE| 144440|
| DELETE| 93130|
| DELETE| 93140|
+----------+--------+我将如何比较这两个数据框架，以便：
如果__change__ = DELETE和masterlist.ttm_id = changeset.ttm_id，那么从Masterlist中删除匹配的ttm_id记录
谢谢!
回答 3
Stack Overflow用户
回答已采纳
发布于 2018-01-26 00:58:04
我喜欢@MaxU使用except的解决方案。下面是另一种使用left_anti联接的方法：
master.join( changelist.where($"__change__" === "DELETE"),
Seq("ttm_id"), "left_anti"
)请注意，对于大型DataFrames，这种方法可能很昂贵。
Stack Overflow用户
发布于 2018-01-25 23:43:36
您可以使用以下查询进行此操作：
select * from masterlist
where not exists (select 1 from changeset
where masterlist.ttm_id = changeset.ttm_id
and masterlist.__change__='DELETE');演示：
scala> m.show
+--------+
| ttm_id|
+--------+
|39622109|
|39622178|
|39578322|
+--------+
scala> c.show
+----------+--------+
|__change__| ttm_id|
+----------+--------+
| DELETE|39622109|
| DELETE| 50520|
+----------+--------+
scala> val q="""
| select * from masterlist
| where not exists (select ttm_id from changeset
| where masterlist.ttm_id = changeset.ttm_id
| and changeset.__change__='DELETE')
| """
q: String =
"
select * from masterlist
where not exists (select ttm_id from changeset
where masterlist.ttm_id = changeset.ttm_id
and changeset.__change__='DELETE')
"
scala> val res = spark.sql(q)
res: org.apache.spark.sql.DataFrame = [ttm_id: int]
scala> res.show
+--------+
| ttm_id|
+--------+
|39622178|
|39578322|
+--------+另一种解决办法是：
scala> import org.apache.spark.sql.functions._
import org.apache.spark.sql.functions._
scala> m.withColumn("__change__", lit("DELETE")).except(c.select("ttm_id","__change__")).select("ttm_id").show
+--------+
| ttm_id|
+--------+
|39578322|
|39622178|
+--------+Stack Overflow用户
发布于 2018-01-26 05:54:45
广播较小的数据帧应该有助于减少加入数据格式所需的洗牌。
广播完join数据后，您可以使用filter和drop来获得您想要的结果。
val broadcastedMasterList = sc.broadcast(changeset)
masterlist.join(broadcastedMasterList.value, Seq("ttm_id"), "left")
.filter($"__change__".isNull || $"__change__" =!= "DELETE")
.drop("__change__")
.show(false)我希望答案是有帮助的。
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/48453281
复制相关文章
相似问题
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号