scala - 根据spark数据帧scala中的列值过滤行

Question

我有一个数据框( Spark ):

id  value 
3     0
3     1
3     0
4     1
4     0
4     0

我想创建一个新的数据框:

3 0
3 1
4 1

需要为每个 id 删除 1(value) 之后的所有行。我尝试在 spark dateframe(Scala) 中使用窗口函数。但无法找到解决方案。似乎是我走错了方向。

我正在 Scala 中寻找解决方案。谢谢

使用 monotonically_increasing_id 输出

 scala> val data = Seq((3,0),(3,1),(3,0),(4,1),(4,0),(4,0)).toDF("id", "value")
data: org.apache.spark.sql.DataFrame = [id: int, value: int]

scala> val minIdx = dataWithIndex.filter($"value" === 1).groupBy($"id").agg(min($"idx")).toDF("r_id", "min_idx")
minIdx: org.apache.spark.sql.DataFrame = [r_id: int, min_idx: bigint]

scala> dataWithIndex.join(minIdx,($"r_id" === $"id") && ($"idx" <= $"min_idx")).select($"id", $"value").show
+---+-----+
| id|value|
+---+-----+
|  3|    0|
|  3|    1|
|  4|    1|
+---+-----+

如果我们在原始数据框中进行排序转换，该解决方案将不起作用。那个时候的 monotonically_increasing_id() 是基于原始 DF 而不是排序的 DF 生成的。我以前错过了那个要求。

欢迎提出所有建议。

Answer 1

一种方法是使用 monotonically_increasing_id()和一个自连接:

val data = Seq((3,0),(3,1),(3,0),(4,1),(4,0),(4,0)).toDF("id", "value")
data.show
+---+-----+
| id|value|
+---+-----+
|  3|    0|
|  3|    1|
|  3|    0|
|  4|    1|
|  4|    0|
|  4|    0|
+---+-----+

现在我们生成一个名为 idx 的列随着 Long :

val dataWithIndex = data.withColumn("idx", monotonically_increasing_id())
// dataWithIndex.cache()

现在我们得到了 min(idx)每个 id哪里 value = 1 :

val minIdx = dataWithIndex
               .filter($"value" === 1)
               .groupBy($"id")
               .agg(min($"idx"))
               .toDF("r_id", "min_idx")

现在我们加入 min(idx)回到原来的 DataFrame :

dataWithIndex.join(
  minIdx,
  ($"r_id" === $"id") && ($"idx" <= $"min_idx")
).select($"id", $"value").show
+---+-----+
| id|value|
+---+-----+
|  3|    0|
|  3|    1|
|  4|    1|
+---+-----+

注: monotonically_increasing_id()根据行的分区生成其值。这个值可能每次都会改变 dataWithIndex被重新评估。在我上面的代码中，由于懒惰的评估，只有当我调用最后的 show那个 monotonically_increasing_id()被评估。

如果您想强制该值保持不变，例如，您可以使用 show要逐步评估上述内容，请取消注释上面的这一行:

//  dataWithIndex.cache()