如何在Databricks中将文件名、文件大小及读取时间写入另一张表

核心逻辑不需要改动你现有文件校验的核心代码，只需要在逐个处理文件的环节中新增元数据采集、写入逻辑即可，具体操作步骤如下： #

1. 确认目标表schema匹配 #

提前保证你已经创建的目标表包含对应字段，以下是参考schema（已建表且字段符合要求可直接跳过）：

CREATE TABLE IF NOT EXISTS file_process_log (
  file_name STRING COMMENT '读取的文件路径/名称',
  file_size BIGINT COMMENT '文件大小，单位为字节',
  read_time TIMESTAMP COMMENT '文件读取执行时间'
) USING DELTA;

如果目标表还有其他自定义字段，只要对应字段允许为空、或设置了默认值，不会影响后续写入逻辑。 #

2. 嵌入元数据采集写入逻辑 #

在你现有逐个处理文件的循环中新增以下代码，Python示例如下：

from pyspark.sql.functions import lit
from datetime import datetime

# 替换为你现有逻辑中获取的待处理文件列表
file_list = ["abfss://xxx@xxx.dfs.core.windows.net/path/file1.csv", "dbfs:/mnt/xxx/file2.parquet"]

for file_path in file_list:
    # --------------------------
    # 此处保留你原有文件读取、校验、执行预设操作的全部代码，不需要改动
    # 示例：df = spark.read.format("csv").load(file_path) + 你自己的校验逻辑
    # --------------------------

    # 新增：采集所需的三个元数据字段
    file_info = dbutils.fs.ls(file_path)[0] # Databricks内置工具直接获取文件信息，无需额外安装依赖
    file_size = file_info.size
    read_time = datetime.now()

    # 生成单行元数据DataFrame，字段顺序、类型和目标表保持一致
    log_df = spark.createDataFrame([(file_path, file_size, read_time)], 
                                   schema=["file_name", "file_size", "read_time"])

    # 追加写入到已创建的目标表
    log_df.write.mode("append").saveAsTable("file_process_log")

3. 大文件量批量写入优化（可选） #

如果单次处理的文件数量超过100个，推荐先把所有元数据攒成列表，全部文件处理完成后统一写入，减少IO开销：

log_list = []
for file_path in file_list:
    # 原有文件处理逻辑不变
    # 元数据采集后追加到列表
    log_list.append((file_path, file_size, read_time))

# 全部文件处理完成后统一写入目标表
log_df = spark.createDataFrame(log_list, schema=["file_name", "file_size", "read_time"])
log_df.write.mode("append").saveAsTable("file_process_log")

如果你用Scala开发，逻辑完全一致，仅需调整对应语法即可。 #

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