Pyspark中使用or筛选多时间周期报错及多周期结果获取方法

报错核心原因 #

• 触发报错的直接原因：PySpark 构建DataFrame过滤逻辑时，不支持用Python原生的and/or/not关键字连接列级判断条件——这些关键字会尝试将Spark的Column对象强制转换为Python原生布尔值，才会抛出你遇到的Cannot convert column into bool错误。列级逻辑或的正确连接符是|，与是&，非是~，且每个独立判断条件必须用括号包裹，避免运算符优先级混乱。
• 之前输出仅包含第一组周期的本质原因：用Python原生or时，语法逻辑会优先取第一个Column表达式的返回值，相当于第二个周期的判断被完全跳过，过滤逻辑等价于只保留第一组周期，自然不会输出后两组的结果。

推荐修改方案 #

针对你需要匹配3组固定周期的场景，优先使用isin()方法做过滤，比写一长串逻辑或判断更简洁、易维护，也不容易出错：

1. 提前拼接好所有需要统计的周期字符串为列表
2. 直接用isin一次性过滤所有目标周期
3. 原有分箱、分组统计逻辑完全保留，分组维度包含period，自然会输出每个周期对应的统计结果，和你期望的输出格式完全匹配

修改后可直接运行的代码：

# 第一步：提前拼接3组目标周期字符串，和原有代码拼接的格式完全一致
target_periods = [
    f"{start} - {end}",
    f"{start_1} - {end_1}",
    f"{start_2} - {end_2}"
]

items_bins = (
  fact_pre
  .filter(f.col('is_client_plus') == 1)
  # 一次性过滤所有指定周期，替代原来的多条件or判断
  .filter(f.col('period').isin(target_periods))
  # 原有分箱逻辑保持不变
  .withColumn('Sum_Spending', f.round('trx_detail_total_amount', 2))
  .withColumn('Sum_Spending', f.when(f.col('Sum_Spending') >= 6000, 6000).otherwise(f.col('Sum_Spending')))
  .withColumn('Sum_Spending_Bin', f.floor(f.col('Sum_Spending') / 250) * 250)
  # 按分箱值、周期分组统计，自动输出每个周期的独立结果
  .groupby('Sum_Spending_Bin', 'period')
  .agg(f.countDistinct('customer_id').alias('LP_Identified_Customers'))
  # 可选：加上排序逻辑，让输出顺序和你给的示例完全一致
  .orderBy('period', 'Sum_Spending_Bin')
)

display(items_bins)

补充说明 #

如果后续遇到多条件复杂判断，必须用逻辑或连接而非isin，正确的写法格式如下（对比原代码的错误写法）：

# 错误写法（原代码）：使用Python原生or
.filter(f.col('period') == 条件1 or f.col('period') == 条件2)

# 正确写法：每个条件块加括号，用|做逻辑或连接
.filter(
  (f.col('period') == f.concat(f.lit(start), f.lit(' - '), f.lit(end)))
  | (f.col('period') == f.concat(f.lit(start_1), f.lit(' - '), f.lit(end_1)))
  | (f.col('period') == f.concat(f.lit(start_2), f.lit(' - '), f.lit(end_2)))
)

这种写法在条件数量多的时候可读性很差，固定值匹配的场景优先用isin更高效。

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