Python遍历文件目录批量预处理Avro数据的实现咨询

核心修改思路 #

把单文件处理逻辑封装为可复用函数，通过遍历目录实现批量处理，同时从文件路径中提取日期、受试者ID、时间戳等元数据，生成唯一的输出文件名避免覆盖。

完整代码实现 #

from avro.datafile import DataFileReader
from avro.io import DatumReader
import csv
import os

def process_avro_file(avro_path, output_dir):
    # 从文件路径提取元数据：日期、受试者ID、时间标签
    dirname, filename = os.path.split(avro_path)
    sub_dir, sub_id = os.path.split(dirname)
    _, date = os.path.split(sub_dir)
    time_tag = os.path.splitext(filename)[0]  # 提取Time1、Time2这类标识
    
    # 读取Avro文件内容
    reader = DataFileReader(open(avro_path, "rb"), DatumReader())
    data = next(reader)  # 假设单个Avro文件仅含一条记录，多记录则改为循环遍历
    reader.close()
    
    # 处理加速度计数据生成时间戳序列
    acc = data["rawData"]["accelerometer"]
    timestamp = [
        round(acc["timestampStart"] + i * (1e6 / acc["samplingFrequency"]))
        for i in range(len(acc["x"]))
    ]
    
    # 生成带元数据的输出文件名
    output_filename = f"accelerometer_{date}_{sub_id}_{time_tag}.csv"
    output_path = os.path.join(output_dir, output_filename)
    
    # 写入CSV文件
    with open(output_path, 'w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow(["unix_timestamp", "x", "y", "z"])
        writer.writerows(zip(timestamp, acc["x"], acc["y"], acc["z"]))
    
    print(f"处理完成：{avro_path} -> {output_path}")

def batch_process_avro(root_dir, output_dir):
    # 确保输出目录存在，不存在则创建
    os.makedirs(output_dir, exist_ok=True)
    
    # 递归遍历根目录下所有Avro文件
    for dirpath, _, filenames in os.walk(root_dir):
        for filename in filenames:
            if filename.endswith(".avro"):
                avro_file_path = os.path.join(dirpath, filename)
                process_avro_file(avro_file_path, output_dir)

# 执行批量处理
if __name__ == "__main__":
    ROOT_DATA_DIR = "/Users/ymhs/projects/"  # 包含所有日期文件夹的根目录
    OUTPUT_DIR = "/Users/ymhs/projects/preproc/"
    batch_process_avro(ROOT_DATA_DIR, OUTPUT_DIR)

关键细节说明 #

1. 路径元数据提取：通过os.path.split逐层拆分文件路径，自动获取日期（如01-01-2024）、受试者ID（如sub-001）、时间标签（如Time1），无需手动输入。
2. 批量遍历逻辑：使用os.walk递归遍历根目录下所有子文件夹，自动筛选.avro后缀的文件进行处理。
3. 输出文件唯一性：生成的CSV文件名包含所有元数据（如accelerometer_01-01-2024_sub-001_Time1.csv），避免不同文件的输出结果互相覆盖。
4. 容错处理：用os.makedirs(output_dir, exist_ok=True)确保输出目录存在，避免因目录未创建导致的报错。

扩展提示 #

• 如果Avro文件包含多条记录，将data = next(reader)改为循环遍历：

  for data in reader:
      # 重复处理每条数据的逻辑
  

• 如需处理其他传感器数据（如陀螺仪），可在process_avro_file中添加类似逻辑，生成对应命名的CSV文件（如gyroscope_{date}_{sub_id}_{time_tag}.csv）。

内容的提问来源于stack exchange，提问作者Linda Jasmine Hoffman