代码之家 › 专栏 › 技术社区 › addmeaning

如何在Spark 2.3.0中减去两个保持重复的数据帧

apache-spark-sql apache-spark

addmeaning · 技术社区 · 7 年前

Spark 2.4.0引入了新的便捷功能 exceptAll 它允许减去两个数据帧,保留重复的数据帧。

实例

  val df1 = Seq(
    ("a", 1L),
    ("a", 1L),
    ("a", 1L),
    ("b", 2L)
  ).toDF("id", "value")
  val df2 = Seq(
    ("a", 1L),
    ("b", 2L)
  ).toDF("id", "value")

df1.exceptAll(df2).collect()
// will return

Seq(("a", 1L),("a", 1L))

但是我只能使用Spark 2.3.0。

仅使用Spark 2.3.0中的函数实现此功能的最佳方法是什么?

0 回复 | 直到 7 年前

Vamsi Prabhala 7 年前

一种选择是使用 row_number 生成序列号列并在 left join 以获取缺少的行。

PySpark解决方案如图所示。

 from pyspark.sql.functions import row_number
 from pyspark.sql import Window
 w1 = Window.partitionBy(df1.id).orderBy(df1.value)
 w2 = Window.partitionBy(df2.id).orderBy(df2.value)
 df1 = df1.withColumn("rnum", row_number().over(w1))
 df2 = df2.withColumn("rnum", row_number().over(w2))
 res_like_exceptAll = df1.join(df2, (df1.id==df2.id) & (df1.val == df2.val) & (df1.rnum == df2.rnum), 'left') \
                         .filter(df2.id.isNull()) \ #Identifies missing rows 
                         .select(df1.id,df1.value)
 res_like_exceptAll.show()

推荐文章

maximodesousadias · 如何根据条件删除日期后的记录

2 年前

Joe Bloggr · 如何将Dataframe类型的函数参数传递给SparkSQL查询

2 年前

Shankar Panda · 如何从org.apache.spark.sql获取密钥。在scala中键入列并将其放入列表变量中?

2 年前

Aaron Brazier · 连接2个pyspark数据帧并继续运行窗口sum和max

2 年前

user23358051 · 火花顺序优化规则

2 年前

Tristpost · 如何从我自己的Java Stream将CSV数据最好地加载到Apache Spark数据帧中?

2 年前

David Cunningham · Pyspark结构化流媒体-来自以前记录的数据

2 年前

Surender Raja · 对case语句的pyspark查询引发错误

2 年前

ConfusedDeveloper · 取消查看SPARK SQL中的列组

3 年前

nfsp412 · 当我使用spark-sql时,将出现此错误

3 年前