将 Spark DataFrame 列中的数字计数拆分为多列

Question

假设我有一个这样的 spark DataFrame

+------------------+----------+--------------+-----+
|              user|        dt|        action|count|
+------------------+----------+--------------+-----+
|Albert            |2018-03-24|Action1       |   19|
|Albert            |2018-03-25|Action1       |    1|
|Albert            |2018-03-26|Action1       |    6|
|Barack            |2018-03-26|Action2       |    3|
|Barack            |2018-03-26|Action3       |    1|
|Donald            |2018-03-26|Action3       |   29|
|Hillary           |2018-03-24|Action1       |    4|
|Hillary           |2018-03-26|Action2       |    2|

我想在单独的计数中对 Action1/Action2/Action3 进行计数，因此将其转换为另一个像这样的 DataFrame

+------------------+----------+-------------+-------------+-------------+
|              user|        dt|action1_count|action2_count|action3_count|
+------------------+----------+-------------+-------------+-------------+
|Albert            |2018-03-24|           19|            0|            0|
|Albert            |2018-03-25|            1|            0|            0|
|Albert            |2018-03-26|            6|            0|            0|
|Barack            |2018-03-26|            0|            3|            0|
|Barack            |2018-03-26|            0|            0|            1|
|Donald            |2018-03-26|            0|            0|           29|
|Hillary           |2018-03-24|            4|            0|            0|
|Hillary           |2018-03-26|            0|            2|            0|

由于我是 Spark 的新手，我尝试实现它是非常枯燥和直接的：

通过每个 "action"
将原始 DF 与每个新 DF 合并，使用新 DF"count" 中的第二个 DF

我试过的代码是这样的：

val a1 = originalDf.filter("action = 'Action1'")
val df1 = originalDf.as('o)
  .join(a1,
        ($"o.user" === $"a1.user" && $"o.dt" === $"a1.dt"), 
        "left_outer")
  .select($"o.user", $"o.dt", $"a1.count".as("action1_count"))

然后对 Action2/Action3 做同样的事情，然后加入这些。

然而，即使在这个阶段，我已经遇到了这种方法的几个问题：

它根本不起作用 - 我的意思是失败并出现错误，我不明白其原因：org.apache.spark.sql.AnalysisException: cannot resolve 'o.user' given input columns: [user, dt, action, count, user, dt, action, count];
即使它成功了，我假设我会在需要零的地方得到空值。
我觉得应该有更好的方法来达到这个目的。像一些地图构造之类的。但目前我觉得我无法构建将第一个数据帧转换为第二个数据帧所需的转换。

所以现在我根本没有可行的解决方案，我将非常感谢您的任何建议。

UPD：我可能还会得到不包含所有 3 个可能的 "action" 值的 DF，例如

+------------------+----------+--------------+-----+
|              user|        dt|        action|count|
+------------------+----------+--------------+-----+
|Albert            |2018-03-24|Action1       |   19|
|Albert            |2018-03-25|Action1       |    1|
|Albert            |2018-03-26|Action1       |    6|
|Hillary           |2018-03-24|Action1       |    4|

对于那些，我仍然需要包含 3 列的结果 DF：

+------------------+----------+-------------+-------------+-------------+
|              user|        dt|action1_count|action2_count|action3_count|
+------------------+----------+-------------+-------------+-------------+
|Albert            |2018-03-24|           19|            0|            0|
|Albert            |2018-03-25|            1|            0|            0|
|Albert            |2018-03-26|            6|            0|            0|
|Hillary           |2018-03-24|            4|            0|            0|

Answer 1

您可以通过使用 when 到 select 适当的列值来避免多个 join。关于你的 join，我真的不认为它有像 cannot resolve 'o.user' 这样的异常，你可能想再次检查你的代码。

val df = Seq(("Albert","2018-03-24","Action1",19),
("Albert","2018-03-25","Action1",1),
("Albert","2018-03-26","Action1",6),
("Barack","2018-03-26","Action2",3),
("Barack","2018-03-26","Action3",1),
("Donald","2018-03-26","Action3",29),
("Hillary","2018-03-24","Action1",4),
("Hillary","2018-03-26","Action2",2)).toDF("user", "dt", "action", "count")

val df2 = df.withColumn("count1", when($"action" === "Action1", $"count").otherwise(lit(0))).
withColumn("count2", when($"action" === "Action2", $"count").otherwise(lit(0))).
withColumn("count3", when($"action" === "Action3", $"count").otherwise(lit(0)))

+-------+----------+-------+-----+------+------+------+
|user   |dt        |action |count|count1|count2|count3|
+-------+----------+-------+-----+------+------+------+
|Albert |2018-03-24|Action1|19   |19    |0     |0     |
|Albert |2018-03-25|Action1|1    |1     |0     |0     |
|Albert |2018-03-26|Action1|6    |6     |0     |0     |
|Barack |2018-03-26|Action2|3    |0     |3     |0     |
|Barack |2018-03-26|Action3|1    |0     |0     |1     |
|Donald |2018-03-26|Action3|29   |0     |0     |29    |
|Hillary|2018-03-24|Action1|4    |4     |0     |0     |
|Hillary|2018-03-26|Action2|2    |0     |2     |0     |
+-------+----------+-------+-----+------+------+------+

Answer 2

这是一种使用 pivot 和 first 的方法，其优点是不必知道 action 值是什么：

val df = Seq(
  ("Albert", "2018-03-24", "Action1", 19),
  ("Albert", "2018-03-25", "Action1", 1),
  ("Albert", "2018-03-26", "Action1", 6),
  ("Barack", "2018-03-26", "Action2", 3),
  ("Barack", "2018-03-26", "Action3", 1),
  ("Donald", "2018-03-26", "Action3", 29),
  ("Hillary", "2018-03-24", "Action1", 4),
  ("Hillary", "2018-03-26", "Action2", 2)
).toDF("user", "dt", "action", "count")

val pivotDF = df.groupBy("user", "dt", "action").pivot("action").agg(first($"count")).
  na.fill(0).
  orderBy("user", "dt", "action")

// +-------+----------+-------+-------+-------+-------+
// |   user|        dt| action|Action1|Action2|Action3|
// +-------+----------+-------+-------+-------+-------+
// | Albert|2018-03-24|Action1|     19|      0|      0|
// | Albert|2018-03-25|Action1|      1|      0|      0|
// | Albert|2018-03-26|Action1|      6|      0|      0|
// | Barack|2018-03-26|Action2|      0|      3|      0|
// | Barack|2018-03-26|Action3|      0|      0|      1|
// | Donald|2018-03-26|Action3|      0|      0|     29|
// |Hillary|2018-03-24|Action1|      4|      0|      0|
// |Hillary|2018-03-26|Action2|      0|      2|      0|
// +-------+----------+-------+-------+-------+-------+

[更新]

根据评论，如果要创建的列的 Action? 多于数据透视列中的列，则可以遍历 missing Action? 以将它们添加为 zero-filled 作为列：

val fullActionList = List("Action1", "Action2", "Action3", "Action4", "Action5")

val missingActions = fullActionList.diff(
  pivotDF.select($"action").as[String].collect.toList.distinct
)
// missingActions: List[String] = List(Action4, Action5)

missingActions.foldLeft( pivotDF )( _.withColumn(_, lit(0)) ).
show

// +-------+----------+-------+-------+-------+-------+-------+-------+
// |   user|        dt| action|Action1|Action2|Action3|Action4|Action5|
// +-------+----------+-------+-------+-------+-------+-------+-------+
// | Albert|2018-03-24|Action1|     19|      0|      0|      0|      0|
// | Albert|2018-03-25|Action1|      1|      0|      0|      0|      0|
// | Albert|2018-03-26|Action1|      6|      0|      0|      0|      0|
// | Barack|2018-03-26|Action2|      0|      3|      0|      0|      0|
// | Barack|2018-03-26|Action3|      0|      0|      1|      0|      0|
// | Donald|2018-03-26|Action3|      0|      0|     29|      0|      0|
// |Hillary|2018-03-24|Action1|      4|      0|      0|      0|      0|
// |Hillary|2018-03-26|Action2|      0|      2|      0|      0|      0|
// +-------+----------+-------+-------+-------+-------+-------+-------+

将 Spark DataFrame 列中的数字计数拆分为多列

Split numerical count in Spark DataFrame column into several columns

dataframe

apache-spark

apache-spark-sql

spark-dataframe