
本文介绍如何使用 PySpark(结合 dbutils.fs)编写可复用函数,根据文件最后修改时间(如 5 天前、7 天前)筛选并移动符合日期条件的文件,支持自定义源路径、目标路径和保留天数参数。
本文介绍如何使用 pyspark(结合 `dbutils.fs`)编写可复用函数,根据文件最后修改时间(如 5 天前、7 天前)筛选并移动符合日期条件的文件,支持自定义源路径、目标路径和保留天数参数。
在 Databricks 环境中使用 PySpark 时,dbutils.fs 是操作 DBFS(Databricks File System)的核心工具。但需注意:dbutils.fs.ls() 返回的 FileInfo 对象不包含文件最后修改时间(modificationTime)字段(截至 Databricks Runtime 14.3+,该字段仍为 None 或不可靠),因此不能直接依赖 file.modificationTime 进行时间筛选。
正确做法是:利用 dbutils.fs.ls() 获取文件元数据后,结合 dbutils.fs.head() 或 dbutils.fs.stat()(推荐)获取精确修改时间。dbutils.fs.stat(path) 返回包含 modificationTime(毫秒级时间戳)的完整文件统计信息,这才是可靠依据。
以下是一个健壮、生产就绪的 Python 函数,支持按“N 天前”条件移动文件:
import time
from datetime import datetime, timedelta
def move_files_by_age(
source_path: str,
dest_path: str,
days_old: int = 7,
dry_run: bool = False
):
"""
将 source_path 下所有修改时间早于 'days_old' 天的文件移动到 dest_path。
Args:
source_path (str): 源路径(DBFS 路径,如 'dbfs:/mnt/raw/abc/')
dest_path (str): 目标路径(DBFS 路径,如 'dbfs:/mnt/archive/abc/')
days_old (int): 文件修改时间需早于当前时间此天数(正数表示“更旧”,如 7 表示 7 天前及更早)
dry_run (bool): 若为 True,则仅打印将要移动的文件,不执行实际操作
Returns:
dict: 包含成功移动数、跳过数、失败数的统计信息
"""
# 计算截止时间戳(毫秒)
cutoff_timestamp_ms = int((datetime.now() - timedelta(days=days_old)).timestamp() * 1000)
files = dbutils.fs.ls(source_path)
moved_count = 0
skipped_count = 0
failed_count = 0
print(f"? 扫描路径: {source_path}")
print(f"? 移动条件: 修改时间 ≤ {datetime.fromtimestamp(cutoff_timestamp_ms / 1000).strftime('%Y-%m-%d %H:%M:%S')} ({days_old} 天前)")
for file_info in files:
# 跳过目录(仅处理文件)
if file_info.isDir:
continue
try:
# 使用 dbutils.fs.stat 获取准确 modificationTime(单位:毫秒)
stat_info = dbutils.fs.stat(file_info.path)
file_mod_time_ms = stat_info.modificationTime
if file_mod_time_ms <p>? <strong>关键说明与注意事项:</strong> </p>
-
时间精度保障:务必使用
dbutils.fs.stat(path)获取modificationTime,而非依赖ls()结果中的不可靠字段; -
路径格式:
source_path和dest_path必须是完整 DBFS 路径(以dbfs:/开头),且dest_path末尾无需加/(函数内部已处理); -
移动 vs 复制:上述代码先
cp后rm实现真正“移动”。若只需复制保留源文件,删除dbutils.fs.rm(...)行即可; -
文件名模式无关性:本方案基于系统修改时间,完全绕开
abc_20250134220000等命名规则,避免因文件重命名或写入延迟导致逻辑错误; -
性能提示:对海量小文件,
stat()调用会有一定开销;如需极致性能且文件名严格遵循时间戳(如abc_20250131220000),可改用正则解析 +datetime比较,但需确保业务写入时间与文件名一致。
通过该函数,您可灵活调度任务(如每日凌晨运行 move_files_by_age(..., days_old=7) 归档一周前数据),实现自动化生命周期管理。










