如何将含NOT EXISTS运算符的SQL查询改写为Spark API代码？

你的错误在于误用了Spark的exists函数——这个函数是针对数组类型的，用于检查数组内是否存在满足lambda条件的元素，和SQL中NOT EXISTS子查询的逻辑完全不同，因此会触发类型不匹配异常。

以下是两种正确的Spark API改写方式，完全匹配原SQL的逻辑：

方案一：左反连接（推荐，性能最优） #

左反连接是Spark实现NOT EXISTS逻辑的标准方式，会返回左表中在右表无匹配的行，完全对应原SQL的查询逻辑：

// 准备关联用的右表，仅保留object_id并别名
Dataset<Row> rightTable = load1.select(functions.col("object_id").alias("match_id"));

// 左反连接，筛选parent_id无对应object_id的行
Dataset<Row> result = load1.join(
        rightTable,
        load1.col("parent_id").equalTo(rightTable.col("match_id")),
        "left_anti"
    )
    .select(
        functions.col("object_id"),
        functions.col("parent_id"),
        functions.col("object_id").alias("root_id"),
        functions.col("attr_id"),
        functions.col("attr_name"),
        functions.col("value")
    );

result.show(false);

方案二：子查询过滤 #

先提取所有存在的object_id集合，再过滤parent_id不在该集合内的行（注意：大数据量下不推荐，因为会将数据拉取到Driver端）：

// 获取所有已存在的object_id列表
List<String> existingIds = load1.select("object_id")
    .collectAsList()
    .stream()
    .map(row -> row.getString(0))
    .toList();

// 筛选parent_id不在existingIds中的行
Dataset<Row> result = load1.filter(
        functions.not(functions.col("parent_id").isin(existingIds.toArray(new String[0])))
    )
    .select(
        functions.col("object_id"),
        functions.col("parent_id"),
        functions.col("object_id").alias("root_id"),
        functions.col("attr_id"),
        functions.col("attr_name"),
        functions.col("value")
    );

result.show(false);

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