本文介绍如何使用 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 <= cutoff_timestamp_ms:
dest_full_path = f"{dest_path.rstrip('/')}/{file_info.name}"
if dry_run:
print(f"[DRY RUN] ✅ 将移动: {file_info.path} → {dest_full_path}")
moved_count += 1
else:
dbutils.fs.cp(file_info.path, dest_full_path, recurse=False)
# 可选：删除源文件（实现「移动」语义，而非复制）
dbutils.fs.rm(file_info.path, recurse=False)
print(f"✅ 已移动: {file_info.name}")
moved_count += 1
else:
print(f"⚠️ 跳过（未达年龄要求）: {file_info.name} (修改于 {datetime.fromtimestamp(file_mod_time_ms/1000).strftime('%Y-%m-%d %H:%M')})")
skipped_count += 1
except Exception as e:
print(f"❌ 移动失败 {file_info.path}: {str(e)}")
failed_count += 1
print(f"\n? 总结: 成功 {moved_count} 个，跳过 {skipped_count} 个，失败 {failed_count} 个")
return {"moved": moved_count, "skipped": skipped_count, "failed": failed_count}
# ✅ 使用示例（移动 5 天前的所有文件）
result = move_files_by_age(
source_path="dbfs:/mnt/raw/abc/",
dest_path="dbfs:/mnt/archive/abc/",
days_old=5,
dry_run=False # 设为 True 先预览
)? 关键说明与注意事项：
-
时间精度保障：务必使用
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) 归档一周前数据），实现自动化生命周期管理。
