构建数据集时FileSystem缺失files_dir属性的AttributeError问题解决

为什么构建阶段没有files_dir属性？ #

• 预览模式是本地调试场景：平台会把输入数据临时挂载到本地文件系统，files_dir就是这个本地挂载目录的路径，方便你用本地文件路径的方式操作。
• 构建阶段是分布式执行场景：数据实际存储在对象存储（比如S3、GCS这类），没有本地文件目录的概念，此时的FileSystem对象是直接对接对象存储的客户端，自然不存在files_dir这个本地路径属性。

解决方案 #

别再用拼接files_dir路径的方式读写文件，改用FileSystem提供的原生API来操作：

1. 推荐：用FileSystem的open方法直接操作文件 #

这种方式不需要依赖本地路径，适配分布式构建场景：

@transform(
    data_input = Input("..."),
    data_output= Output("..."),
)
def transform(data_input, data_output):
    fs_input = data_input.filesystem()
    fs_output = data_output.filesystem()
    
    for file_status in fs_input.files('*.<file_ending>').collect():
        # 读取输入文件：用open方法获取文件对象
        with fs_input.open(file_status.path, 'r') as input_file:
            # 根据PreProcessor的实际逻辑调整，比如传入文件对象或读取内容
            data = PreProcessor(input_file)
        
        # 写入输出文件：同样用open方法创建输出文件对象
        with fs_output.open("transformed_file", 'w') as output_file:
            # 调整PreProcessor的write方法，让它接受文件对象而非本地路径
            data.write(output_file)

2. 如果PreProcessor必须用本地文件路径 #

如果你的PreProcessor只能接受本地文件路径，那就先把对象存储的文件下载到临时目录，处理完再上传：

import tempfile
import os

@transform(
    data_input = Input("..."),
    data_output= Output("..."),
)
def transform(data_input, data_output):
    fs_input = data_input.filesystem()
    fs_output = data_output.filesystem()
    
    for file_status in fs_input.files('*.<file_ending>').collect():
        # 创建临时输入文件，下载对象存储的文件到本地
        with tempfile.NamedTemporaryFile(delete=False) as tmp_in:
            fs_input.download(file_status.path, tmp_in.name)
        
        # 用临时路径处理数据
        data = PreProcessor(tmp_in.name)
        
        # 创建临时输出文件，写入处理后的数据
        with tempfile.NamedTemporaryFile(delete=False) as tmp_out:
            data.write(tmp_out.name)
        
        # 上传临时文件到输出的对象存储
        fs_output.upload(tmp_out.name, "transformed_file")
        
        # 清理临时文件
        os.unlink(tmp_in.name)
        os.unlink(tmp_out.name)

3. 简化方案：用Output的内置写入方法 #

部分平台的Output对象提供了更便捷的写入方法，比如直接写入内容：

@transform(
    data_input = Input("..."),
    data_output= Output("..."),
)
def transform(data_input, data_output):
    fs_input = data_input.filesystem()
    
    for file_status in fs_input.files('*.<file_ending>').collect():
        with fs_input.open(file_status.path, 'r') as f:
            # 假设PreProcessor有process方法直接处理内容并返回结果
            processed_content = PreProcessor.process(f.read())
        
        # 直接写入输出，无需手动操作FileSystem
        data_output.write_data(processed_content, file_name="transformed_file")

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