在 PySpark 中，agg() 内部无法直接引用同一 agg() 中刚通过 .alias() 定义的新列名（如 "business_count"），因为列别名在当前聚合阶段尚未进入逻辑计划的可用列集；必须通过链式转换（如 withColumn）或重复调用聚合函数来实现跨列计算。
在 pyspark 中，`agg()` 内部无法直接引用同一 `agg()` 中刚通过 `.alias()` 定义的新列名（如 `"business_count"`），因为列别名在当前聚合阶段尚未进入逻辑计划的可用列集；必须通过链式转换（如 `withcolumn`）或重复调用聚合函数来实现跨列计算。
PySpark 的 agg() 方法（以及 select、withColumn 等列操作）遵循列作用域隔离原则：所有传入 agg() 的表达式会在同一执行阶段并行计算，且仅能访问输入 DataFrame 的原始列（或已存在的列），不能引用本 agg() 中其他表达式所定义的别名。这与 SQL 中 SELECT 子句的语义完全一致——你无法在同一个 SELECT 里用 col2 = col1 + 1 后立即写 col3 = col2 * 2。
例如，以下代码会报错：
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
spark = SparkSession.builder.getOrCreate()
df = spark.range(3)
# ❌ 报错：AnalysisException: Column 'x' does not exist
df.select(
(F.col("id") * 2).alias("x"),
F.col("x") + 10 # 尝试引用同级 alias → 失败
).show()
回到你的原始需求：统计高分（4/5 星）商家数量，并计算其占总评论数的百分比。以下是两种推荐方案：
✅ 方案一：链式 withColumn（推荐，清晰可维护）
total_reviews = reviews_df.count()
r = reviews_df.alias("r")
result = (
r.filter(r.review_stars.isin([4, 5]))
.agg(F.count("business_id").alias("business_count"))
.withColumn(
"top_rated_pct",
F.round(F.col("business_count") / total_reviews * 100, 0)
)
)
result.show()
# +----------------+---------------+
# |business_count|top_rated_pct|
# +----------------+---------------+
# | 3| 60|
# +----------------+---------------+
✅ 方案二：复用聚合函数（避免别名依赖）
result = (
r.filter(r.review_stars.isin([4, 5]))
.agg(
F.count("business_id").alias("business_count"),
F.round(F.count("business_id") / total_reviews * 100, 0).alias("top_rated_pct")
)
)
⚠️ 注意事项：
total_reviews是驱动端变量（Python int），不是列，因此可安全用于withColumn或agg中的标量计算；withColumn链式调用在 Spark Catalyst 优化器中通常会被合并为单个物理执行计划，不会带来额外性能开销；- 若需多层派生列（如
pct,level = when(pct > 50, "high")），务必使用连续withColumn，而非试图在单次select/agg中嵌套引用。
总之，牢记一个准则：“定义即不可见” —— 在 agg() 或 select() 中定义的别名，只在该操作返回的 DataFrame 中生效，无法被同一操作内的其他表达式消费。拆解为多步、语义明确的转换，既是 Spark 最佳实践，也更利于调试与协作。
