用id对该框架进行分区，创建一个布尔标志，以检查event_name和event_value在任何行中为（'A'，0）时的条件，然后使用此标志沿着提供的其他条件对该框架进行filter。

exp = F.expr("event_name = 'A' AND event_value = 0").cast('int')
counts = F.sum(exp).over(Window.partitionBy('id'))

mask = ~((F.col('event_value') == 0) & (counts > 0)) | (F.col('event_name') == 'A')
df1 = df.withColumn('mask', mask).filter('mask')

+---+----------+-----------+----+
| id|event_name|event_value|mask|
+---+----------+-----------+----+
|  1|         A|          0|true|
|  1|         B|          2|true|
|  2|         A|          5|true|
|  2|         B|          0|true|
|  2|         C|          2|true|
|  3|         A|          7|true|
|  3|         B|          8|true|
|  3|         C|          9|true|
+---+----------+-----------+----+

使用window函数，然后根据需要使用filter。请检查下面的代码

scala> spark.table("input").show(false)
+---+----------+-----------+
|id |event_name|event_value|
+---+----------+-----------+
|1  |A         |0          |
|1  |B         |2          |
|1  |C         |0          |
|2  |A         |5          |
|2  |B         |0          |
|2  |C         |2          |
|3  |A         |7          |
|3  |B         |8          |
|3  |C         |9          |
+---+----------+-----------+

scala> :paste
// Entering paste mode (ctrl-D to finish)

spark.sql("""
    WITH wn_input AS (
    SELECT
        *,
        (RANK() OVER(PARTITION BY ID ORDER BY event_name, event_value)) AS rid
    FROM input
)
SELECT
    id,
    event_name AS name,
    event_value AS value
FROM wn_input WHERE not ( rid = 1 AND event_name = 'A' AND event_value = 0)
""").show(false)

// Exiting paste mode, now interpreting.

+---+----+-----+
|id |name|value|
+---+----+-----+
|1  |B   |2    |
|1  |C   |0    |
|2  |A   |5    |
|2  |B   |0    |
|2  |C   |2    |
|3  |A   |7    |
|3  |B   |8    |
|3  |C   |9    |
+---+----+-----+