Azure Databricks处理大文件时命令挂起问题求助

我遇到过类似的Databricks大文件批量处理挂起的情况，结合你的代码和描述，问题大概率出在资源未完全释放或DBFS文件系统交互的隐性锁/缓存上。以下是几个针对性的排查和解决建议：

1. 强制清理临时文件与确保文件句柄释放 #

你的代码在处理完文件后没有清理DBFS临时目录的文件，虽然内存占用稳定，但Databricks的/dbfs映射层可能存在文件缓存或残留句柄，累积后导致后续任务卡住。

调整代码： #

在每个文件处理完成后，添加临时文件删除逻辑：

# 处理完文件后立即清理临时文件
dbutils.fs.rm(f'dbfs:/tmp/temporary/{file}')
dbutils.fs.rm(f'dbfs:/tmp/temporary/{xml_filename}')

同时，在批量写入后强制刷新文件缓冲区，避免数据滞留在内存缓存中：

if len(to_write) > 10_000:
    outfile.write(''.join(to_write))
    outfile.flush()  # 强制将数据刷入磁盘
    to_write = []
# 最后一次写入也需要刷新
outfile.write(''.join(to_write))
outfile.flush()

2. 替换DBFS临时目录为本地临时文件 #

Databricks的/dbfs是FUSE映射的分布式文件系统，大文件的持续读写可能触发底层的锁机制或I/O阻塞。改用本地临时文件（集群节点的本地磁盘）处理，能绕开分布式文件系统的潜在问题。

调整代码： #

import tempfile
import os

text_files = ['file1.txt','file2.txt','file3.txt']
for file in text_files:
    xml_filename = file.replace('.txt','.xml')
    
    # 下载源文件到本地临时文件
    with tempfile.NamedTemporaryFile(delete=False, mode='rb') as local_txt:
        dbutils.fs.cp(
            f'dbfs:/mnt/storage_account/projects/xml_converter/input/{file}',
            f'file:{local_txt.name}'
        )
    
    # 本地处理转换
    with tempfile.NamedTemporaryFile(delete=False, mode='w', encoding='utf-8') as local_xml, \
         open(local_txt.name, 'r') as infile:
        to_write = []
        line_count = 0
        for line in infile:
            # 你的XML转换逻辑
            new_xml = f"<line>{line.strip()}</line>\n"  # 示例转换
            to_write.append(new_xml)
            
            if len(to_write) > 10_000:
                local_xml.write(''.join(to_write))
                local_xml.flush()
                to_write = []
            
            line_count += 1
            if line_count % 1_000_000 == 0:
                print(f"Processed {line_count} lines for {file}")
        
        local_xml.write(''.join(to_write))
        local_xml.flush()
    
    # 上传转换后的文件到输出目录
    dbutils.fs.cp(
        f'file:{local_xml.name}',
        f"/mnt/storage_account/projects/xml_converter/output/{xml_filename}"
    )
    
    # 删除本地临时文件
    os.unlink(local_txt.name)
    os.unlink(local_xml.name)

3. 将单循环拆分为独立Notebook任务 #

Databricks的单个Notebook单元格长时间运行后，可能会累积进程级的状态残留（比如未释放的系统资源）。把单个文件的处理逻辑拆分为独立Notebook，通过dbutils.notebook.run调用，每个任务完成后会自动释放进程资源。

步骤： #

1. 创建一个名为process_single_file的Notebook，添加代码：

# 获取传入的文件名参数
file = dbutils.widgets.get("input_file")
xml_filename = file.replace('.txt','.xml')

# 这里复制你的文件下载、转换、上传、清理逻辑
# ...（和之前的处理代码一致）

2. 在主Notebook中批量调用：

text_files = ['file1.txt','file2.txt','file3.txt']
for file in text_files:
    # 设置超时时间（比如3600秒=1小时）
    dbutils.notebook.run("process_single_file", 3600, {"input_file": file})

4. 添加详细日志定位挂起点 #

如果以上方法还没解决，建议添加详细的日志记录，明确任务卡在哪个阶段（复制、转换、上传）：

import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

for idx, file in enumerate(text_files):
    logger.info(f"=== Starting file {idx+1}/{len(text_files)}: {file} ===")
    
    logger.info("Copying input file to temporary location")
    dbutils.fs.cp(f'dbfs:/mnt/storage_account/projects/xml_converter/input/{file}',f'dbfs:/tmp/temporary/{file}')
    
    logger.info("Starting XML conversion")
    line_count = 0
    with open(f'/dbfs/tmp/temporary/{file}','r') as infile, open(f'/dbfs/tmp/temporary/{xml_filename}','a', encoding="utf-8") as outfile:
        to_write = []
        for line in infile:
            line_count +=1
            # 转换逻辑
            new_xml = f"<line>{line.strip()}</line>\n"
            to_write.append(new_xml)
            
            if len(to_write) > 10_000:
                outfile.write(''.join(to_write))
                outfile.flush()
                to_write = []
            
            if line_count % 1_000_000 == 0:
                logger.info(f"Processed {line_count} lines for {file}")
        
        outfile.write(''.join(to_write))
        outfile.flush()
    logger.info(f"Conversion completed, total lines: {line_count}")
    
    logger.info("Moving output file to storage")
    dbutils.fs.cp(f'dbfs:/tmp/temporary/{xml_filename}',f"/mnt/storage_account/projects/xml_converter/output/{xml_filename}")
    
    logger.info("Cleaning up temporary files")
    dbutils.fs.rm(f'dbfs:/tmp/temporary/{file}')
    dbutils.fs.rm(f'dbfs:/tmp/temporary/{xml_filename}')
    
    logger.info(f"=== Completed processing {file} ===\n")

通过日志可以精准定位是哪个步骤导致的挂起，比如如果卡在Moving output file，那大概率是DBFS的I/O锁问题。

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