本文介绍如何使用 PySpark（结合 dbutils.fs）编写可复用函数，根据文件最后修改时间（如 5 天或 7 天前），将符合时间条件的文件从源路径批量移动到目标路径，并支持按命名规则（如 abc_20250134220000）灵活筛选。
本文介绍如何使用 pyspark（结合 dbutils.fs）编写可复用函数，根据文件最后修改时间（如 5 天或 7 天前），将符合时间条件的文件从源路径批量移动到目标路径，并支持按命名规则（如 abc_20250134220000）灵活筛选。
在 Databricks 环境中，dbutils.fs 是操作 DBFS 文件系统的核心工具，但需注意：dbutils.fs.cp() 是复制操作，而原需求是「移动」（即复制 + 删除源文件）。由于 DBFS 不直接提供原子性 mv 跨目录移动（尤其跨不同存储账户时），标准做法是先 cp 后 rm，确保数据一致性。
以下是一个健壮、参数化、生产就绪的 PySpark 函数，支持：
- 按 最后修改时间（modification time） 筛选（如
days_old=7表示仅移动 7 天前修改的文件）； - 可选按 文件名正则匹配（适配
abc_20250134220000类型命名）； - 自动处理路径拼接与异常防护；
- 返回移动统计信息便于日志追踪。
✅ 完整实现代码
import re
from datetime import datetime, timedelta
from pyspark.sql.functions import col
def move_files_by_age(
source_path: str,
dest_path: str,
days_old: int,
file_pattern: str = None,
dry_run: bool = False
) -> dict:
"""
将 source_path 下满足「修改时间早于 N 天前」且（可选）匹配正则 pattern 的文件，
移动至 dest_path。支持 dry_run 模式预览。
:param source_path: 源 DBFS 路径（如 'dbfs:/mnt/raw/abc/'）
:param dest_path: 目标 DBFS 路径（如 'dbfs:/mnt/archive/abc/'）
:param days_old: 整数，表示「多少天前」（如 7 → 移动 7 天前修改的文件）
:param file_pattern: 可选，正则字符串（如 r'^abc_\d{14}\.csv$'）
:param dry_run: 若为 True，则只打印将要执行的操作，不实际移动
:return: 包含 moved_count、skipped_count、errors 的字典
"""
from datetime import datetime, timedelta
import time
# 计算时间阈值（UTC 时间戳，单位秒）
cutoff_timestamp = time.time() - (days_old * 24 * 60 * 60)
# 列出源路径所有文件（不递归子目录）
try:
files = dbutils.fs.ls(source_path)
except Exception as e:
raise ValueError(f"无法访问源路径 {source_path}: {str(e)}")
moved_count = 0
skipped_count = 0
errors = []
for file_info in files:
# 跳过目录（仅处理文件）
if file_info.isDir:
continue
# 检查文件名模式（若提供）
if file_pattern and not re.match(file_pattern, file_info.name):
skipped_count += 1
continue
# 检查修改时间（dbutils.fs.ls 返回的 modificationTime 是毫秒级时间戳）
if file_info.modificationTime / 1000 < cutoff_timestamp:
src_full = file_info.path
dst_full = f"{dest_path.rstrip('/')}/{file_info.name}"
try:
if not dry_run:
# 先复制（支持跨挂载点）
dbutils.fs.cp(src_full, dst_full)
# 再删除源文件（完成「移动」语义）
dbutils.fs.rm(src_full)
# 日志记录（无论 dry_run）
print(f"[MOVE] {src_full} → {dst_full}")
moved_count += 1
except Exception as e:
error_msg = f"移动失败 {src_full}: {str(e)}"
errors.append(error_msg)
print(f"[ERROR] {error_msg}")
else:
skipped_count += 1
return {
"moved_count": moved_count,
"skipped_count": skipped_count,
"errors": errors,
"summary": f"完成扫描：{len(files)} 个条目，移动 {moved_count} 个，跳过 {skipped_count} 个"
}
# ✅ 使用示例：移动 7 天前的 abc_YYYYMMDDHHMMSS 格式 CSV 文件
result = move_files_by_age(
source_path="dbfs:/mnt/raw/abc/",
dest_path="dbfs:/mnt/archive/abc/",
days_old=7,
file_pattern=r"^abc_\d{14}\.csv$",
dry_run=False # 设为 True 可先试运行
)
print(result["summary"])
if result["errors"]:
print("⚠️ 存在错误:", result["errors"])⚠️ 关键注意事项
-
时间基准：
file_info.modificationTime是 DBFS 中记录的最后修改时间（毫秒级 Unix 时间戳），精确可靠；它不依赖文件名中的日期，因此比解析abc_20250134220000更鲁棒（尤其当文件写入延迟或重命名发生时）。 -
路径格式：务必使用
dbfs:/前缀（而非/dbfs/），因为dbutils.fs只识别 DBFS URI 格式。 -
原子性保障：本函数采用
cp + rm模式。如需强一致性（防中间状态），建议在关键流程中添加事务日志或使用 Delta Lake 的VACUUM配合审计表。 -
性能提示：
dbutils.fs.ls()在海量小文件场景下可能较慢；若文件量 >10k，建议改用spark.read.text()扫描元数据目录（需启用 DBFS 日志），或预聚合到 Delta 表。 -
权限检查：确保当前 notebook 的集群拥有源路径的
READ和WRITE权限，以及目标路径的WRITE权限。
? 总结
该函数将原始脚本升级为工业级工具：参数清晰、错误可控、逻辑可验证、扩展性强。你可根据业务需要轻松调整 days_old（如设为 5 即移动 5 天前文件），并利用正则精准约束文件范围，避免误移。首次运行建议开启 dry_run=True 进行沙盒验证，确认逻辑无误后再执行真实移动。
