pyspark:自动填充隐式缺失值

Question

我有一个数据框

user day amount
a 2 10
a 1 14
a 4 5
b 1 4

你看，day的最大值是4，最小值是1。amount我要填0列在所有用户的所有缺失日期中，因此上述数据框将变为。

user day amount
    a 2 10
    a 1 14
    a 4 5
    a 3 0
    b 1 4
    b 2 0
    b 3 0
    b 4 0

我如何在 PySpark 中做到这一点？非常感谢。

Answer 1

这是一种方法。您可以先获取最小值和最大值，然后对 user 列和数据透视进行分组，然后填充缺失的列并将所有空值填充为 0，然后将它们堆叠回去:

min_max = df.agg(F.min("day"),F.max("day")).collect()[0]
df1 = df.groupBy("user").pivot("day").agg(F.first("amount").alias("amount")).na.fill(0)

missing_cols = [F.lit(0).alias(str(i)) for i in range(min_max[0],min_max[1]+1) 
                                                if str(i) not in df1.columns ]
df1 = df1.select("*",*missing_cols)

#+----+---+---+---+---+
#|user|  1|  2|  4|  3|
#+----+---+---+---+---+
#|   b|  4|  0|  0|  0|
#|   a| 14| 10|  5|  0|
#+----+---+---+---+---+

#the next step is inspired from https://stackoverflow.com/a/37865645/9840637
arr = F.explode(F.array([F.struct(F.lit(c).alias("day"), F.col(c).alias("amount"))
                                           for c in df1.columns[1:]])).alias("kvs")
(df1.select(["user"] + [arr])
    .select(["user"]+ ["kvs.day", "kvs.amount"]).orderBy("user")).show()

---

+----+---+------+
|user|day|amount|
+----+---+------+
|   a|  1|    14|
|   a|  2|    10|
|   a|  4|     5|
|   a|  3|     0|
|   b|  1|     4|
|   b|  2|     0|
|   b|  4|     0|
|   b|  3|     0|
+----+---+------+

请注意，由于列 day 已旋转，数据类型可能已更改，因此您可能必须将它们转换回原始数据类型