缺少 spark-dataframe 枢轴 columns/values

spark-dataframe pivot missing columns/values

我现在遇到了一个我无法解决的问题,让我解释一下。

我需要对 spark-dataframe 进行数据透视,但在某些情况下,数据透视没有任何记录来包含我需要的列。这是一个例子:

假设有 4 种类型的事件:A、B、C、D

我收到这样的事件日志文件:

|Id|year|type|
|--|----|----|
|a |2015|   A|
|a |2016|   A|
|a |2015|   A|
|a |2017|   A|
|b |2015|   A|
|b |2015|   B|
|b |2016|   D|
|b |2015|   B|
|b |2017|   A|  

当我做枢轴时,我得到:

|id|year|   A|   B|   D|
|--|----|----|----|----|
|a |2015|   2|null|null|
|a |2016|   1|null|null|
|a |2017|   1|null|null|
|b |2015|   1|   2|null|
|b |2016|   1|null|   1|
|b |2017|   1|null|null|

我真正需要的是:

|id|year|   A|   B|   C|   D|
|--|----|----|----|----|----|
|a |2015|   2|   0|   0|   0|
|a |2016|   1|   0|   0|   0|
|a |2017|   1|   0|   0|   0|
|b |2015|   1|   2|   0|   0|
|b |2016|   1|   0|   0|   1|
|b |2017|   1|   0|   0|   0|

想象一下,在我收到的每个事件日志文件中都缺少不同类型的事件,但我始终需要拥有所有事件类型(A、B、C 和 D)的列。

我希望我解释得很好。

val lscol = Seq((“A”),(“B”),(“C”),(“D”)).toDF(“Type”)

//df is the initial dataframe prior to pivot
df.join(lscol,df.col(“Type”)===lscol.col(“Type”),”right”)
    .drop(df(“Type”))
    .groupBy(“ID”,”year”)
    .pivot(“Type”)
    .agg(count(“Type”))
    .filter(!col(“ID”).isNull && !col(“year”).isNull)
    .na
    .fill(0)
    .show