使用DeltaTable关联两张表实现数据删除的问题排查

业务场景 #

存在一个包含company_id和date字段的S3源文件，另一张S3目标Delta表包含company_id、date、year等多列，需根据源表的company_id和date匹配，删除目标表中年份≤2010的数据。

原代码及报错 #

尝试代码 #

val source="s3a://source_table"
val target="s3a://target_table"

val source_delta = DeltaTable.forPath(spark, source)
val target_delta = DeltaTable.forPath(spark, target)

target_delta .alias("a")
      .merge(
        source_delta .alias("b"),
        condition=( ($"b.company_id" === $"a.company_id") and  ($"b.data" === $"a.date") and ($"a.year" <= "2010") )
        )
      .whenMatched().delete()
      .execute()

println("Done")

报错信息 #

command-4296412619890660:18: error: overloaded method value merge with alternatives:
  (source: org.apache.spark.sql.DataFrame,condition: org.apache.spark.sql.Column)io.delta.tables.DeltaMergeBuilder <and>
  (source: org.apache.spark.sql.DataFrame,condition: String)io.delta.tables.DeltaMergeBuilder
 cannot be applied to (io.delta.tables.DeltaTable, condition: org.apache.spark.sql.Column)
      .merge(
       ^

报错原因 #

1. 参数类型不匹配：DeltaTable的merge方法要求第一个参数必须是DataFrame，但代码中传入的是DeltaTable对象，导致类型不兼容。
2. 字段名笔误：代码中$"b.data"应为$"b.date"，源表字段名写错。
3. 逻辑位置错误：年份过滤条件不应放在merge的匹配条件中，否则会导致仅匹配年份≤2010的行，逻辑不符合需求。

修正后的代码 #

val source = "s3a://source_table"
val target = "s3a://target_table"

val source_delta = DeltaTable.forPath(spark, source)
val target_delta = DeltaTable.forPath(spark, target)

target_delta.alias("a")
  .merge(
    // 将DeltaTable转为DataFrame，满足merge方法参数要求
    source_delta.toDF().alias("b"),
    // 仅用company_id和date作为匹配条件
    $"b.company_id" === $"a.company_id" && $"b.date" === $"a.date"
  )
  // 匹配后，仅删除year≤2010的目标行
  .whenMatched($"a.year" <= 2010).delete()
  .execute()

println("Done")

关键修正点 #

• 将source_delta通过toDF()转为DataFrame，符合merge方法的参数类型要求。
• 修正字段名错误：$"b.data"改为$"b.date"。
• 调整条件逻辑：将年份过滤从merge匹配条件移到whenMatched的删除条件中，确保先匹配源和目标的company_id与date，再删除符合年份要求的目标行，逻辑更准确。

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