如何实现基于createTime自动删除5分钟以上数据的Delta表？

一、替代VACUUM定期调度的实时/准实时清理方式 #

1. 借助结构化流的foreachBatch触发清理 #

在写入流的foreachBatch回调中，完成当前批次数据写入后，直接执行删除过期数据的SQL。这种方式可以绑定流的触发频率（比如1分钟一次），实现准实时清理，无需额外调度任务。

示例代码（Scala）：

val streamQuery = df.writeStream
  .format("delta")
  .option("checkpointLocation", "/path/to/checkpoint")
  .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
    // 写入当前批次符合时间要求的数据
    batchDF.write.mode("append").format("delta").save("/path/to/delta-table")
    
    // 清理5分钟前的过期数据
    val fiveMinutesAgo = System.currentTimeMillis() - 5 * 60 * 1000
    spark.sql(s"""
      DELETE FROM delta.`/path/to/delta-table` 
      WHERE createTime < ${fiveMinutesAgo}
    """)
  }
  .start()

2. 利用Delta Live Tables（DLT）的自动约束清理 #

如果使用DLT，可以通过定义流式表+约束规则，让系统自动清理过期数据。DLT会在数据处理环节自动校验约束，删除不符合条件的旧数据。

示例SQL定义：

CREATE OR REFRESH STREAMING TABLE realtime_data_table
AS SELECT * FROM stream_source
WHERE createTime >= current_timestamp() - INTERVAL 5 MINUTE;

-- 添加自动清理约束
ALTER STREAMING TABLE realtime_data_table
ADD CONSTRAINT expire_old_records
CHECK (createTime >= current_timestamp() - INTERVAL 5 MINUTE)
APPLY AS DELETE;

二、基于createTime列自动删除过期数据的Delta表实现 #

Delta表本身没有内置的自动生命周期删除功能，但可以通过以下方式实现类似效果：

1. 表属性标记规则+定时删除 #

先给Delta表添加属性记录过期规则，再配合Spark作业或调度工具（如Airflow）定期执行删除逻辑：

-- 设置表属性记录过期规则
ALTER TABLE my_delta_table
SET TBLPROPERTIES (
  'data.expire.rule' = 'createTime < current_timestamp() - INTERVAL 5 MINUTE'
);

-- 定期执行删除
DELETE FROM my_delta_table
WHERE createTime < current_timestamp() - INTERVAL 5 MINUTE;

2. 开启Change Data Feed（CDF）+ 流监听清理 #

开启Delta表的CDF后，用另一个流监听表的变更，实时检查并删除过期数据：

-- 开启表的CDF功能
ALTER TABLE my_delta_table SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

示例流处理代码（Scala）：

val cdfStream = spark.readStream
  .format("delta")
  .option("readChangeFeed", "true")
  .load("/path/to/delta-table")
  .writeStream
  .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
    val fiveMinutesAgo = current_timestamp() - INTERVAL 5 MINUTE
    spark.sql(s"""
      DELETE FROM delta.`/path/to/delta-table` 
      WHERE createTime < '$fiveMinutesAgo'
    """)
  }
  .start()

注意事项 #

• 上述DELETE操作属于逻辑删除，物理文件仍会保留，需要配合VACUUM定期清理旧文件，但可以降低VACUUM的执行频率（比如每小时一次）以减少资源消耗。
• 如果数据量极大，频繁DELETE会影响性能，建议按时间分桶（比如每分钟一个桶），直接删除整个过期桶的目录，效率更高。

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