火花数据框用其他行的值替换值

Question

我有一个包含很多列的 df，但我的问题是 2 列：

val df = Seq(("id1","unknown"),("id1","type1"),("id1","unknown"),("id2","typeX"),
             ("id2","typeX"),("id2","unknown"),("id5","typeY"),("id2","unknown"))
    .toDF("ID","TYPE")
+---+-------+
| ID|   TYPE|
+---+-------+
|id1|unknown|
|id1|  type1|
|id1|unknown|
|id2|  typeX|
|id2|  typeX|
|id2|unknown|
|id5|  typeY|
|id2|unknown|
+---+-------+

我想将类型“未知”替换为与 ID 对应的类型。 结果应该是这样的：

+---+-----+
| ID| TYPE|
+---+-----+
|id1|type1|
|id1|type1|
|id1|type1|
|id2|typeX|
|id2|typeX|
|id2|typeX|
|id5|typeY|
|id2|typeX|
+---+-----+

它不能被硬编码（使用when id1 -> type1 等），因为我有 300 000 个每周更改的 ID...

这是我已经尝试过的：

val w = Window.partitionBy("ID")

df.withColumn("TYPE",collect_list("TYPE").over(w))

+---+--------------------------------+
|ID |TYPE                            |
+---+--------------------------------+
|id5|[typeY]                         |
|id1|[unknown, type1, unknown]       |
|id1|[unknown, type1, unknown]       |
|id1|[unknown, type1, unknown]       |
|id2|[typeX, typeX, unknown, unknown]|
|id2|[typeX, typeX, unknown, unknown]|
|id2|[typeX, typeX, unknown, unknown]|
|id2|[typeX, typeX, unknown, unknown]|
+---+--------------------------------+

df.withColumn("TYPE",typeProcessingUDF(col("TYPE")))

+---+-----+
| ID| TYPE|
+---+-----+
|id5|typeY|
|id1|type1|
|id1|type1|
|id1|type1|
|id2|typeX|
|id2|typeX|
|id2|typeX|
|id2|typeX|
+---+-----+

def dtypeProcessing(dtypeList : mutable.WrappedArray[String]) : String = {
    val dtype = dtypeList
        .filter(element => element!= "unknown" && element!="")
        .distinct
    dtype.length match {
        case 0 => "Unknown"
        case x if x >1 => "Unknown"
        case x if x ==1 => dtype(0)
    }
}
val typeProcessingUDF = udf(dtypeProcessing _)

这行得通，

但它并没有处理考虑案例的所有情况：

• if [type1,type2] => return "Unknown"
• if [type1,type2,type2] => return type2

Answer 1

窗口为“ID”，函数“first”忽略空值：

val idWindow = Window.partitionBy("ID")
val unknownToNull = when($"TYPE" === "unknown", null).otherwise($"TYPE")
val result = df.withColumn("TYPE",
  coalesce(unknownToNull,
    first(unknownToNull, ignoreNulls = true).over(idWindow)
  )
)

输出：

+---+-----+
|ID |TYPE |
+---+-----+
|id1|type1|
|id1|type1|
|id1|type1|
|id2|typeX|
|id2|typeX|
|id2|typeX|
|id2|typeX|
|id5|typeY|
+---+-----+