PySpark中如何实现Azure Data Flow的countAll与countAllDistinct功能？

实现countAll（包含空值的列计数） #

Azure Data Flow里的countAll(col)会统计指定列对应的所有行（包括null值），PySpark中可以通过count(lit(1))来实现这个效果——因为lit(1)会为每一行生成一个非空的常量值，对它计数就等价于统计该列对应的所有行数，不管原列是否为null。

示例（按分组聚合统计）：

from pyspark.sql import functions as F

# 按id分组，统计value列的所有行数（包含null）
df.groupBy("id").agg(F.count(F.lit(1)).alias("countAll_value")).show()

如果是统计整个DataFrame的总行数（全局countAll），直接用df.count()即可，这个方法本身就会统计所有行，包含存在null的行。

实现countAllDistinct（包含空值的去重计数） #

PySpark的countDistinct(col)会自动忽略null值，要实现包含null的去重计数，有两种可靠的方式：

方法1：将null替换为唯一标识后统计

把null替换成一个不会在原列中出现的特殊值，再用countDistinct统计，这样null会被当作一个有效值计入去重结果。

示例：

from pyspark.sql import functions as F

# 全局统计value列的去重值（包含null）
df.agg(
    F.countDistinct(F.coalesce(F.col("value"), F.lit("__UNIQUE_NULL_MARKER__"))).alias("countAllDistinct_value")
).show()

# 分组聚合场景
df.groupBy("id").agg(
    F.countDistinct(F.coalesce(F.col("value"), F.lit("__UNIQUE_NULL_MARKER__"))).alias("countAllDistinct_value")
).show()

注意：要确保__UNIQUE_NULL_MARKER__这个值不会出现在你的业务数据中，避免统计错误。

方法2：结合countDistinct和null存在性判断

先统计非null值的去重数量，再判断原列是否存在null，存在则加1（因为所有null去重后只算一个）。

示例（兼容PySpark全版本）：

from pyspark.sql import functions as F

df.agg(
    (F.countDistinct("value") + 
     F.when(F.count(F.when(F.col("value").isNull(), 1)) > 0, 1).otherwise(0)
    ).alias("countAllDistinct_value")
).show()

如果使用PySpark 3.0及以上版本，也可以用exists简化null判断：

df.agg(
    (F.countDistinct("value") + 
     F.when(F.exists(F.col("value"), lambda x: x.isNull()), 1).otherwise(0)
    ).alias("countAllDistinct_value")
).show()

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