如何循环遍历数据集以创建摘要数据集答案

【问题标题】：How to loop through dataset to create a dataset of summary如何循环遍历数据集以创建摘要数据集
【发布时间】：2018-07-17 07:21:39
【问题描述】：

我刚开始学习和使用 Spark，目前面临一个问题。任何建议或提示将不胜感激。

基本上我有一个数据集，其中包含不同用户的各种事件，例如 AppLaunch、GameStart、GameEnd 等，我想为每个用户在每次启动应用程序时的操作创建一个摘要。

例如：我有以下数据集：
UserId | Event Type | Time | GameType | Event Id|
11111 | AppLauch | 11:01:53| null | 101 |
11111 | GameStart | 11:01:59| Puzzle | 102 |
11111 | GameEnd | 11:05:31| Puzzle | 103 |
11111 | GameStart | 11:05:58| Word | 104 |
11111 | GameEnd | 11:09:13| Word | 105 |
11111 | AppEnd | 11:09:24| null | 106 |
11111 | AppLauch | 12:03:43| null | 107 |
22222 | AppLauch | 12:03:52| null | 108 |
22222 | GameStart | 12:03:59| Puzzle | 109 |
11111 | GameStart | 12:04:01| Puzzle | 110 |
22222 | GameEnd | 12:06:11| Puzzle | 111 |
11111 | GameEnd | 12:06:13| Puzzle | 112 |
11111 | AppEnd | 12:06:23| null | 113 |
@ 987654335@

只需要知道第一次玩的游戏和最后一次玩的游戏，即使在一次应用启动中玩了 3 场以上的游戏。

我最初的想法是按用户ID和时间范围窗口（AppLaunch到AppEnd）对它们进行分组，然后找到一种方法扫描数据集，如果有一个gameStart事件并且它落入任何窗口，它将是 FirstGamePlayed，AppEnd 时间之前的最后一个 GameStart 事件将是 LastGamePlayed。但我没有找到实现这一目标的方法。

任何提示/建议都会很好。

谢谢

【问题讨论】：

首先使用窗口函数获取第一个和最后一个游戏，然后 groupBy 减少到每个 userId 1 行
但是那个 groupBy userId 不就分不清是什么事件了吗？

标签： apache-spark spark-dataframe

【解决方案1】：

我认为这可以使用窗口函数和如下聚合来解决：

df
   // enumerate AppLaunches 
   .withColumn("AppLauchNr", sum(when($"EventType" === "AppLauch", 1)).over(Window.partitionBy($"UserId").orderBy($"Time".asc)))
   // get first last game per AppLaunch
   .withColumn("firstGamePlayed", first($"GameType", true).over(Window.partitionBy($"UserId", $"AppLauchNr").orderBy($"Time".asc)))
   .withColumn("lastGamePlayed", first($"GameType", true).over(Window.partitionBy($"UserId", $"AppLauchNr").orderBy($"Time".desc)))
    // now aggregate
   .groupBy($"AppLauchNr")
   .agg(
        first($"UserId").as("UserId"),
        min($"EventId").as("EventId"),
        lit("AppLauch").as("EventType"), // this is always AppLauch
        min($"Time").as("Time"),
        first($"firstGamePlayed", true).as("firstGamePlayed"),
        first($"lastGamePlayed", true).as("lastGamePlayed")
   )
  .drop($"AppLauchNr")

第一场和最后一场比赛也可以使用orderBy().groupBy()而不是窗口函数来确定，但我仍然不确定spark在聚合过程中是否保留了排序（这在文档中没有提到，参见例如Spark DataFrame: does groupBy after orderBy maintain that order?和在https://issues.apache.org/jira/browse/SPARK-16207讨论）

 df
   .withColumn("AppLauchNr", sum(when($"EventType" === "AppLauch", 1)).over(Window.partitionBy($"UserId").orderBy($"Time".asc)))
   .orderBy($"UserId",$"AppLauchNr",$"Time")
   .groupBy($"UserId",$"AppLauchNr")
   .agg(
        first($"EventId").as("EventId"),
        first($"EventType").as("EventType"),
        first($"Time").as("Time"),
        first($"GameType", true).as("firstGamePlayed"),
        last($"GameType", true).as("lastGamePlayed")
   )
   .drop($"AppLauchNr")

【讨论】：