在 pyspark 中执行 partitionBy 列时消除特定列的空值行

Question

我有一个这样的 pyspark 数据框：

+-----+---+-----+
| id| name|state|
+-----+---+-----+
|111| null|   CT|
|222|name1|   CT|
|222|name2|   CT|
|333|name3|   CT|
|333|name4|   CT|
|333| null|   CT|
+---+-----+-----+

对于给定的 ID，我想保留该记录，即使列“名称”为空，如果它的 ID 不重复，但如果 ID 重复，那么我想检查名称列并确保它在该 ID 中不包含重复项，并且如果“名称”为空，则仅针对重复的 ID 删除。以下是所需的输出：

+-----+---+-----+
| id| name|state|
+-----+---+-----+
|111| null|   CT|
|222|name1|   CT|
|222|name2|   CT|
|333|name3|   CT|
|333|name4|   CT|
+---+-----+-----+

如何在 PySpark 中实现这一点？

Answer 1

您可以通过按 id 列分组并计算每个组中的名称数量来做到这一点。默认情况下，Spark 中将忽略 Null 值，因此应保留任何计数为 0 的组。我们现在可以过滤掉组中计数大于 0 的任何空值。

在 Scala 中，这可以通过如下窗口函数来完成：

val w = Window.partitionBy("id")
val df2 = df.withColumn("gCount", count($"name").over(w))
  .filter($"name".isNotNull or $"gCount" === 0)
  .drop("gCount")

PySpark 等价物：

w = Window.partitionBy("id")
df.withColumn("gCount", count("name").over(w))
  .filter((col("name").isNotNull()) | (col("gCount") == 0))
  .drop("gCount")

---

以上内容不会删除同一 id 具有多个 null 的行（所有这些都将保留）。

如果这些也应该被删除，只保留一行name==null，一个简单的方法是在运行上述代码之前或之后使用.dropDuplicates(['id','name'])。请注意，这也会删除任何其他重复项（在这种情况下，.dropDuplicates(['id','name', 'state']) 可能更可取）。

Answer 2

我认为您可以分两步完成。首先，按id计数值

import pyspark.sql.window as psw
w = psw.Window.partitionBy("id")
df = df.withColumn("n",psf.sum(psf.lit(1)).over(w))

然后过滤去除Null当n<1:

df.filter(!((psf.col('name').isNull()) & (psf.col('n') > 1)))

编辑

正如@Shubham Jain 所述，如果name 有多个Null 值（重复），则上述过滤器将保留它们。在这种情况下，@Shaido 提出的解决方案很有用：使用.dropDuplicates(['id','name']) 添加后处理。或.dropDuplicates(['id','name','state'])，根据您的喜好