PySpark导入CSV时如何排除空白行标记原生null值行

问题原因 #

默认读取CSV时，Spark会自动将配置的nullValue（默认为空值）、emptyValue统一转换为DataFrame的null类型，导入完成后，原始CSV里的空字符串""、自定义null标记:null已经被转成相同的null值，无法通过isNull()方法区分二者。

实现步骤 #

你需要在读取CSV阶段保留原始文本值，先做原生null标记，再做常规的空值转换和类型映射，具体实现如下：

1. 读取CSV时禁用自动空值转换

from pyspark.sql import SparkSession
import pyspark.sql.functions as F

spark = SparkSession.builder.appName("null_mark").getOrCreate()

# 读取时保留所有原始字符串，不自动转null
df_raw = spark.read \
    .option("header", "true") \
    .option("inferSchema", "false") \
    .option("emptyValue", "") \
    .option("nullValue", "_NO_MATCH_") \ # 填入一个不存在的标记，避免自动识别任意值为null
    .csv("some_people.csv")

2. 标记原生null后再做常规数据清洗

df = df_raw \
    # 标记原生null行，判断条件对应你CSV中实际的null标记值，示例中为`:null`
    .withColumn("status", F.when(F.col("age") == ":null", 1).otherwise(0)) \
    # 按业务规则将空字符串转成null，做正常的类型转换
    .withColumn("first_name", F.when(F.col("first_name") == "", None).otherwise(F.col("first_name"))) \
    .withColumn("age", F.when(F.col("age").isin("", ":null"), None).otherwise(F.col("age").cast("int")))

如果需要判断多列的原生null，可以按列扩展when的判断条件即可。

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