PySpark使用.select时莫名新增含NULL行的问题排查

核心错误点 #

你代码里的lit(category_name)用错了——lit()是用来生成常量列的，比如lit("apple")会给每行填充字符串"apple"。如果你想把分组列category_name的实际值复制到新列categoryName，应该用col(category_name)来引用原列的值，而不是lit()。

额外NULL/空串行的原因 #

虽然你确认原列无NULL和空串，但大概率是验证不够彻底：

• 数值类型列的NaN不会被isNull()检测到，但Spark的groupBy会把NaN当成独立分组；
• 字符串列的空格、不可见字符（比如制表符）会被视为有效分组，容易被误判为空串；
• 可能你验证时没做trim()处理，导致带空格的字符串漏检。

修正后的代码 #

from pyspark.sql.types import IntegerType  # 根据实际字段类型调整，比如StringType

dfCalls = spark.read.table('foo.bar.my_table') \
                .groupBy(category_name) \
                .count() \
                .withColumn('categoryName', col(category_name))  # 正确引用原列值
                .withColumn('dimCategoryId', lit(None).cast(IntegerType()))  # 显式指定类型，避免类型推断问题
                .select('dimCategoryId', 'categoryName', col(category_name).alias('categoryValue')) \
                .alias('cm')

彻底验证原表数据 #

先排查原表是否存在隐藏的异常值：

original_df = spark.read.table('foo.bar.my_table')

# 检查NULL值
print("NULL行数:", original_df.filter(col(category_name).isNull()).count())

# 检查空串及空格
print("空串/空格行数:", original_df.filter(trim(col(category_name)) == "").count())

# 检查数值列的NaN
col_type = dict(original_df.dtypes)[category_name]
if col_type in ['double', 'float']:
    print("NaN行数:", original_df.filter(col(category_name).isNaN()).count())

分组结果验证 #

在select前先查看分组后的结果，确认是否存在异常分组：

grouped_df = spark.read.table('foo.bar.my_table').groupBy(category_name).count()
grouped_df.show(truncate=False)
# 过滤疑似异常值
grouped_df.filter(col(category_name).isNull() | (trim(col(category_name)) == "")).show()

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