Apache Spark中含空值/空字符串的两个数据集对比问题

第一步：统一空值格式 #

先把数据中各种形式的空值（"null"、"NULL"、空格、空字符串）统一转换为Spark原生的null，避免因空值格式差异导致对比误判。可以在读取DataFrame后添加清洗逻辑：

// 注册空值归一化UDF
sparkSession.udf().register("normalizeNull", (String s) -> {
    if (s == null) return null;
    String trimmed = s.trim();
    if (trimmed.isEmpty() || trimmed.equalsIgnoreCase("null")) {
        return null;
    }
    return trimmed;
}, DataTypes.StringType);

// 清洗test列并重新注册临时视图
mainDF.createOrReplaceTempView("main");
compareDF.createOrReplaceTempView("compare");
Dataset<Row> cleanedMainDF = sparkSession.sql("SELECT *, normalizeNull(test) as cleaned_test FROM main").drop("test").withColumnRenamed("cleaned_test", "test");
Dataset<Row> cleanedCompareDF = sparkSession.sql("SELECT *, normalizeNull(test) as cleaned_test FROM compare").drop("test").withColumnRenamed("cleaned_test", "test");
cleanedMainDF.createOrReplaceTempView("main");
cleanedCompareDF.createOrReplaceTempView("compare");

第二步：修改SQL筛选逻辑 #

原SQL仅筛选两边存在缺失的行，现在调整为只保留test列值不同的行，同时排除两边test均为空的情况。修改后的SQL如下：

SELECT * 
FROM (SELECT 'main' AS main_flag, main.* FROM main) main
NATURAL FULL JOIN (SELECT 'compare' AS compare_flag, compare.* FROM compare) compare
WHERE 
    -- 排除两边test都为空的行
    NOT (main.test IS NULL AND compare.test IS NULL)
    -- 筛选test列值不同的场景：一方有值一方为空，或两边值不相等
    AND (main.test <> compare.test OR main.test IS NULL OR compare.test IS NULL)

完整调整后的代码片段 #

将上述逻辑整合到原代码中：

final SparkSession sparkSession=SparkSession.builder().appName("Final Project").master("local[3]").getOrCreate();
final DataFrameReader reader = sparkSession.read();
reader.option("header", "true");
Dataset<Row> mainDF = reader.csv(mainFile);
Dataset<Row> compareDF = reader.csv(compareFile);

// 注册空值归一化UDF
sparkSession.udf().register("normalizeNull", (String s) -> {
    if (s == null) return null;
    String trimmed = s.trim();
    if (trimmed.isEmpty() || trimmed.equalsIgnoreCase("null")) {
        return null;
    }
    return trimmed;
}, DataTypes.StringType);

// 清洗test列并重新注册视图
mainDF.createOrReplaceTempView("main");
compareDF.createOrReplaceTempView("compare");
Dataset<Row> cleanedMainDF = sparkSession.sql("SELECT *, normalizeNull(test) as cleaned_test FROM main").drop("test").withColumnRenamed("cleaned_test", "test");
Dataset<Row> cleanedCompareDF = sparkSession.sql("SELECT *, normalizeNull(test) as cleaned_test FROM compare").drop("test").withColumnRenamed("cleaned_test", "test");
cleanedMainDF.createOrReplaceTempView("main");
cleanedCompareDF.createOrReplaceTempView("compare");

// 执行修改后的对比SQL
Dataset<Row> joinDF = sparkSession.sql("SELECT * FROM (SELECT 'main' AS main_flag, main.* FROM main) main NATURAL FULL JOIN (SELECT 'compare' AS compare_flag, compare.* FROM compare) compare WHERE NOT (main.test IS NULL AND compare.test IS NULL) AND (main.test <> compare.test OR main.test IS NULL OR compare.test IS NULL)");

joinDF.coalesce(1).write().mode("overwrite").
        format("csv").option("header", "true").
        save("src/main/resources/comparePrototype/Test");

逻辑说明 #

1. 空值归一化：把所有表示空的字符串统一转为Spark原生null，确保不同格式的空值在对比时被视为相同值。
2. SQL条件优化：
   • NOT (main.test IS NULL AND compare.test IS NULL)：直接过滤掉两边test列都为空的无效差异行。
   • (main.test <> compare.test OR main.test IS NULL OR compare.test IS NULL)：覆盖所有test列值不同的有效场景，包括一方为空一方有值、两边值不相等的情况。

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