如何比较多行？答案

【问题标题】：How to compare multiple rows?如何比较多行？
【发布时间】：2015-11-15 20:44:29
【问题描述】：

我想比较两个连续的行 i 与 i-1 的 col2（按 col1 排序）。

如果i-th 行的item_i 和item_[i-1]_row 不同，我想将item_[i-1] 的计数加1。

+--------------+
| col1 col2    |
+--------------+
| row_1 item_1 |
| row_2 item_1 |
| row_3 item_2 |
| row_4 item_1 |
| row_5 item_2 |
| row_6 item_1 |
+--------------+

在上面的例子中，如果我们一次向下扫描两行，我们看到row_2 和row_3 是不同的，因此我们在 item_1 中添加一个。接下来，我们看到row_3与row_4不同，然后在item_2上加一。继续，直到我们结束：

+-------------+
|  col2  col3 |
+-------------+
|  item_1  2  |
|  item_2  2  |
+-------------+

【问题讨论】：

标签： scala apache-spark spark-streaming apache-spark-sql

【解决方案1】：

您可以结合使用窗口函数和聚合来执行此操作。窗口函数用于获取col2 的下一个值（使用col1 进行排序）。然后聚合计算我们遇到差异的次数。这在下面的代码中实现：

val data = Seq(
  ("row_1", "item_1"),
  ("row_2", "item_1"),
  ("row_3", "item_2"),
  ("row_4", "item_1"),
  ("row_5", "item_2"),
  ("row_6", "item_1")).toDF("col1", "col2")

import org.apache.spark.sql.expressions.Window
val q = data.
  withColumn("col2_next",
    coalesce(lead($"col2", 1) over Window.orderBy($"col1"), $"col2")).
  groupBy($"col2").
  agg(sum($"col2" =!= $"col2_next" cast "int") as "col3")

scala> q.show
17/08/22 10:15:53 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.
+------+----+
|  col2|col3|
+------+----+
|item_1|   2|
|item_2|   2|
+------+----+

【讨论】：