在 Spark 中使用 groupBy 并返回到 DataFrame

Question

我在使用 Scala 处理 spark 中的数据帧时遇到了困难。如果我有一个数据框，我想提取一列唯一条目，当我使用 groupBy 时，我没有得到数据框。

例如，我有一个名为 logs 的 DataFrame，其格式如下：

machine_id  | event     | other_stuff
 34131231   | thing     |   stuff
 83423984   | notathing | notstuff
 34131231   | thing    | morestuff

我想要唯一的机器 ID，其中事件是存储在新 DataFrame 中的东西，以便我可以进行某种过滤。使用

val machineId = logs
  .where($"event" === "thing")
  .select("machine_id")
  .groupBy("machine_id")

我得到了一组分组数据，使用起来很麻烦（或者我不知道如何正确使用这种对象）。获得此唯一机器 ID 列表后，我想使用它来过滤另一个 DataFrame 以提取单个机器 ID 的所有事件。

我可以看出我想经常做这种事情，基本工作流程是：

这是前两个步骤，希望能在这里提供一些指导。

我很欣赏这个例子有点做作，但希望它能解释我的问题。可能是我对 GroupedData 对象了解不够，或者（正如我所希望的那样）我在数据框中遗漏了一些使这变得容易的东西。我正在使用基于 Scala 2.10.4 构建的 spark 1.5。

谢谢

Answer 1

只需使用 distinct 而不是 groupBy:

val machineId = logs.where($"event"==="thing").select("machine_id").distinct

相当于SQL:

SELECT DISTINCT machine_id FROM logs WHERE event = 'thing'

GroupedData 不打算直接使用。它提供了很多方法，其中agg是最通用的，可用于应用不同的聚合函数并将其转换回DataFrame。就 SQL 而言，你在 where 和 groupBy 之后的内容相当于这样的东西

SELECT machine_id, ... FROM logs WHERE event = 'thing' GROUP BY machine_id

其中 ... 必须由 agg 或等效方法提供。

Answer 2

在 spark 中分组，然后聚合，然后 select 语句将 return 一个数据框。对于您的示例，它应该类似于：

val machineId = logs
    .groupBy("machine_id", "event")
    .agg(max("other_stuff") )
    .select($"machine_id").where($"event" === "thing")

Using groupBy in Spark and getting back to a DataFrame