SQL转PySpark查询结果不一致：EXISTS用法困惑

仅用于学习，非生产场景。

原始SQL查询及输出 #

SELECT count(*) 
FROM orders_bronze_view 
WHERE o_orderdate >= date("1993-07-01") 
  AND o_orderdate < date("1993-10-01") 
  AND EXISTS ( 
    SELECT * 
    FROM lineitem_bronze_view 
    WHERE l_orderkey = o_orderkey 
      AND l_commitdate < l_receiptdate 
  )

输出：263458

注：orders_bronze_view对应DataFrameorders_bronze，lineitem_bronze_view对应DataFramelineitem_bronze，已确认DataFrame和视图计数一致。

我写的PySpark代码及输出 #

(orders_bronze 
 .join(lineitem_bronze, orders_bronze.o_orderkey == lineitem_bronze.l_orderkey, "left") 
 .filter( 
    (col("l_commitdate") < col("l_receiptdate")) & 
    (col("o_orderdate") >= lit("1993-07-01")) & 
    (col("o_orderdate") < lit("1993-10-1")) 
 ) 
 .count() 
)

输出：726795

问题核心 #

SQL里的EXISTS逻辑是只要存在至少一条满足条件的关联记录，就将当前order计入统计，不管对应多少条lineitem；而你用left join后再过滤的写法，会把每个匹配的lineitem都保留，一个order如果对应N条符合条件的lineitem，就会被重复计数N次，这就是结果偏大的原因。

正确的PySpark写法 #

方法1：贴合SQL逻辑的exists子查询写法 #

from pyspark.sql import functions as F

# 先筛选出符合条件的lineitem子集
lineitem_filtered = lineitem_bronze.filter(F.col("l_commitdate") < F.col("l_receiptdate"))

# 对orders应用exists条件并计数
result = (orders_bronze
          .filter(
              (F.col("o_orderdate") >= F.lit("1993-07-01")) &
              (F.col("o_orderdate") < F.lit("1993-10-01")) &
              F.exists(
                  lineitem_filtered,
                  lambda li: li.l_orderkey == F.col("o_orderkey")
              )
          )
          .count()
)
print(result)

方法2：先去重有效orderkey再关联 #

from pyspark.sql import functions as F

# 获取所有满足条件的唯一orderkey
valid_order_keys = (lineitem_bronze
                   .filter(F.col("l_commitdate") < F.col("l_receiptdate"))
                   .select("l_orderkey")
                   .distinct())

# 关联orders并统计符合时间条件的记录数
result = (orders_bronze
          .filter(
              (F.col("o_orderdate") >= F.lit("1993-07-01")) &
              (F.col("o_orderdate") < F.lit("1993-10-01"))
          )
          .join(valid_order_keys, orders_bronze.o_orderkey == valid_order_keys.l_orderkey, "inner")
          .count())
print(result)

方法3：分组后判断是否存在有效记录 #

from pyspark.sql import functions as F

result = (orders_bronze
          .join(lineitem_bronze, orders_bronze.o_orderkey == lineitem_bronze.l_orderkey, "left")
          .filter(
              (F.col("o_orderdate") >= F.lit("1993-07-01")) &
              (F.col("o_orderdate") < F.lit("1993-10-01"))
          )
          .groupBy("o_orderkey")
          .agg(F.max(F.when(F.col("l_commitdate") < F.col("l_receiptdate"), 1).otherwise(0)).alias("has_valid_lineitem"))
          .filter(F.col("has_valid_lineitem") == 1)
          .count())
print(result)

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