PySpark 3.3.0：基于条件聚合求和避免自连接计算失败率

Absolutely, you can absolutely do conditional summation to calculate the failure rate in a single pass without extra DataFrames or joins—this is the efficient way to handle this kind of aggregation!

Your initial attempt was on the right track, but you had the order reversed. Instead of chaining .when() onto sum(), you need to wrap the conditional logic inside the sum() function. The when() function lets you selectively include values in the sum only when the condition (outcome = "failure") is met, otherwise you use 0.

Scala (Spark) Implementation #

Here's a clean, efficient solution:

import org.apache.spark.sql.functions.{sum, when, col}

// Option 1: With intermediate columns for clarity
val failureRateDF = df.groupBy("endPoint")
  .agg(
    sum(when(col("outcome") === "failure", col("count")).otherwise(0)).alias("failureCount"),
    sum("count").alias("totalCount")
  )
  .withColumn("failureRate", col("failureCount") / col("totalCount"))
  .drop("failureCount", "totalCount")

// Option 2: Concise version (compute rate directly in aggregation)
val failureRateDF = df.groupBy("endPoint")
  .agg(
    (sum(when(col("outcome") === "failure", col("count")).otherwise(0)) / sum("count")).alias("failureRate")
  )

PySpark Implementation #

The logic is identical—just adjust the syntax for Python:

from pyspark.sql.functions import sum, when, col

failure_rate_df = df.groupBy("endPoint") \
    .agg(
        (sum(when(col("outcome") == "failure", col("count")).otherwise(0)) / sum("count")).alias("failureRate")
    )

Why This Works #

This approach handles all calculations in one group-by aggregation:

1. We sum only the count values where outcome is "failure"
2. We sum all count values for each endpoint
3. We divide the two sums to get the failure rate

You avoid the overhead of creating a separate failure-only DataFrame and joining it back, which is far more efficient—especially with large datasets.

内容的提问来源于stack exchange，提问作者daniel9x