识别窗口上的列的重复值（Scala）答案

【问题标题】：Identifying recurring values a column over a Window (Scala)识别窗口上的列的重复值（Scala）
【发布时间】：2019-03-21 16:14:25
【问题描述】：

我有一个包含两列的数据框：“ID”和“Amount”，每一行代表特定 ID 的交易和交易金额。我的示例使用以下 DF：

val df = sc.parallelize(Seq((1, 120),(1, 120),(2, 40),
  (2, 50),(1, 30),(2, 120))).toDF("ID","Amount")

我想创建一个新列来标识所述金额是否为经常性值，即是否发生在相同 ID 的任何其他交易中。

我找到了一种更通用的方法，即跨越整个“金额”列，不考虑 ID，使用以下函数：

def recurring_amounts(df: DataFrame, col: String) : DataFrame = {
  var df_to_arr = df.select(col).rdd.map(r => r(0).asInstanceOf[Double]).collect()
  var arr_to_map = df_to_arr.groupBy(identity).mapValues(_.size)
  var map_to_df = arr_to_map.toSeq.toDF(col, "Count")
  var df_reformat = map_to_df.withColumn("Amount", $"Amount".cast(DoubleType))
  var df_out = df.join(df_reformat, Seq("Amount"))
  return df_new
}

val df_output = recurring_amounts(df, "Amount")

这会返回：

+---+------+-----+
|ID |Amount|Count|
+---+------+-----+
| 1 | 120  |  3  |
| 1 | 120  |  3  |
| 2 |  40  |  1  |
| 2 |  50  |  1  | 
| 1 |  30  |  1  |
| 2 | 120  |  3  |
+---+------+-----+

然后我可以使用它来创建我想要的二进制变量，以指示金额是否经常出现（如果 > 1 则为是，否则为否）。

但是，我的问题在此示例中通过值 120 进行了说明，该值在 ID 1 中重复出现，但在 ID 2 中不重复出现。因此，我想要的输出是：

 +---+------+-----+
|ID |Amount|Count|
+---+------+-----+
| 1 | 120  |  2  |
| 1 | 120  |  2  |
| 2 |  40  |  1  |
| 2 |  50  |  1  | 
| 1 |  30  |  1  |
| 2 | 120  |  1  |
+---+------+-----+

我一直在想一种方法来应用一个函数 .over(Window.partitionBy("ID") 但不知道该怎么做。任何提示将不胜感激。

【问题讨论】：

标签： scala apache-spark-sql user-defined-functions window-functions distinct-values

【解决方案1】：

如果你擅长 sql，你可以为你的Dataframe 编写 sql 查询。您需要做的第一件事是将您的 Dataframe 注册为 spark 内存中的表。之后，您可以在表的顶部编写 sql。请注意，spark 是 spark 会话变量。

val df = sc.parallelize(Seq((1, 120),(1, 120),(2, 40),(2, 50),(1, 30),(2, 120))).toDF("ID","Amount")
df.registerTempTable("transactions")
spark.sql("select *,count(*) over(partition by ID,Amount) as Count from transactions").show()

如果您有任何问题，请告诉我。

【讨论】：