基于row_key关联用Spark SQL/Scala筛选表1同键下独有的key对应行

基础场景对齐 #

• 两张表字段完全统一，包含row_key、data_as_of_date、key、value四个字段
• 表关联维度为row_key，单表内同一个row_key可对应多条不同业务记录
• 筛选规则：逐row_key做范围匹配，返回表1中、同row_key范围下key未在表2出现的所有完整数据行
• 预期输出匹配示例：返回row_key=A的key3对应记录，row_key=B的key2、key3对应记录

Spark SQL 实现 #

优先采用Spark引擎做了专项优化的**左反连接（Left Anti Join）**实现，相比子查询NOT IN的写法，不存在数据膨胀、空值匹配失效的问题，执行效率更高。

-- 替换为你实际的表名即可直接执行
SELECT t1.*
FROM table1 t1
LEFT ANTI JOIN table2 t2
  ON t1.row_key = t2.row_key
  AND t1.key = t2.key;

写法说明：

• 连接条件同时绑定row_key和key两个判断维度，等价于在每个row_key的独立分组内逐行核对key是否重合
• 完全匹配预期结果：只要表1的记录在同row_key的表2记录里找不到同key的行，就会被完整保留
• 避坑提示：不要写全局范围的子查询过滤，例如WHERE t1.key NOT IN (SELECT key FROM table2)，这类写法没有按row_key限定匹配范围，会把其他row_key下存在的同key值误判为重合，导致结果漏数。

Scala Spark DataFrame 实现 #

逻辑和SQL层完全对齐，同样基于左反连接算子实现，代码如下：

import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions.col

// 加载两张目标表的DataFrame
val table1: DataFrame = spark.read.table("你的库名.table1")
val table2: DataFrame = spark.read.table("你的库名.table2")

// 定义同row_key下key匹配的连接条件
val joinCond = table1.col("row_key") === table2.col("row_key") &&
               table1.col("key") === table2.col("key")

// 执行左反连接，仅保留表1中匹配不上的记录
val resultDF = table1
  .join(table2, joinCond, "left_anti")
  // 显式选择表1原始字段，保证输出结构和表1完全一致，避免冗余字段混入
  .select(table1.columns.map(col): _*)

实现说明：

• 采用left_anti连接类型，运行时不会像普通Left Join那样产生重复行，也不需要额外做去重、空值过滤操作
• 当表2数据量较小时，Spark会自动触发广播Join优化，大幅降低跨节点shuffle的性能开销
• 输出的DataFrame字段结构和表1完全一致，不需要额外做字段裁剪。

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