Databricks无需大量try-except实现自定义日志持久化方案问询

问题背景 #

我正在为Databricks中的多个作业搭建日志系统，目前用io.StringIO()实现内存日志记录。现在所有代码块都用try-except包裹，用来捕获日志和异常，确保最后上传日志到Blob存储的代码块能执行，但这种写法太繁琐。想知道有没有办法在代码完全报错时仍执行最后代码块，或者有其他方案能确保任何错误发生时日志都能直接上传。

方案一：用try-finally统一管理逻辑与日志上传 #

把所有业务代码集中放在一个try块中，日志上传逻辑放到finally块内。finally块的代码无论try块是否抛出异常都会执行，这样就不用给每个业务代码片段单独加try-except，大幅简化结构。

示例代码

import io
import logging
from pyspark.sql.functions import col

# 日志配置
log_stream = io.StringIO()
logger = logging.getLogger(database_name_bron)
logger.setLevel(logging.DEBUG)

formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler = logging.StreamHandler(log_stream)
handler.setLevel(logging.DEBUG)
handler.setFormatter(formatter)

if logger.hasHandlers():
    logger.handlers.clear()
logger.addHandler(handler)

# 业务逻辑与日志上传
try:
    # 所有业务代码都放在这里
    table = output_dict['*'].select(
        col('1*').alias('1*'),
        col('2*').alias('2*'),
        col('3*').alias('3*'),
        col('4*').alias('4*'),
        col('5*').alias('5*'),
    )

    # 表左连接操作
    table2 = table2.join(table1, table2['5*'] == table1['4*'], 'left')
    logger.info('table1与table2完成左连接')

    # 其他业务步骤...
except Exception as e:
    # 全局捕获异常并记录详细信息
    logger.exception(f"执行过程中发生错误: {e}")
finally:
    # 无论是否出现异常，都会执行日志上传
    log_content = log_stream.getvalue()
    dbutils.fs.put(
        f"abfss://{container_name}@{storage_account}.dfs.core.windows.net/{p_container_name}",
        log_content,
        overwrite=True
    )
    
    # 清理日志处理器资源
    logger.removeHandler(handler)
    handler.close()

方案二：自定义Blob存储日志处理器 #

实现一个自定义的日志Handler，让日志直接写入缓冲区，最后统一上传到Blob存储。这种方式避免了手动维护内存日志流的繁琐，日志产生时自动格式化写入缓冲区，异常场景下也能通过finally确保缓冲区内容被上传。

示例代码

import logging
from pyspark.sql.functions import col
from io import StringIO

class BlobLogHandler(logging.Handler):
    def __init__(self, container_name, storage_account, blob_path):
        super().__init__()
        self.container_name = container_name
        self.storage_account = storage_account
        self.blob_path = blob_path
        self.log_buffer = StringIO()

    def emit(self, record):
        # 格式化日志条目并写入缓冲区
        log_entry = self.format(record)
        self.log_buffer.write(f"{log_entry}\n")

    def flush(self):
        # 将缓冲区内容上传至Blob存储
        log_content = self.log_buffer.getvalue()
        dbutils.fs.put(
            f"abfss://{self.container_name}@{self.storage_account}.dfs.core.windows.net/{self.blob_path}",
            log_content,
            overwrite=True
        )
        # 重置缓冲区
        self.log_buffer.seek(0)
        self.log_buffer.truncate()

# 日志配置
logger = logging.getLogger(database_name_bron)
logger.setLevel(logging.DEBUG)

formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
blob_handler = BlobLogHandler(container_name, storage_account, p_container_name)
blob_handler.setLevel(logging.DEBUG)
blob_handler.setFormatter(formatter)

if logger.hasHandlers():
    logger.handlers.clear()
logger.addHandler(blob_handler)

# 业务逻辑
try:
    table = output_dict['*'].select(
        col('1*').alias('1*'),
        col('2*').alias('2*'),
        col('3*').alias('3*'),
        col('4*').alias('4*'),
        col('5*').alias('5*'),
    )

    table2 = table2.join(table1, table2['5*'] == table1['4*'], 'left')
    logger.info('table1与table2完成左连接')

    # 其他业务步骤...
except Exception as e:
    logger.exception(f"执行过程中发生错误: {e}")
finally:
    # 确保所有日志都被上传
    blob_handler.flush()
    logger.removeHandler(blob_handler)
    blob_handler.close()

方案说明 #

• 方案一的核心是利用finally块的特性，确保日志上传逻辑必然执行，同时通过全局try-except统一捕获异常，避免重复编写冗余的异常处理代码。
• 方案二通过自定义Handler实现日志的自动化管理，日志产生时自动写入缓冲区，既简化了代码结构，也能灵活控制日志上传的时机（比如可以在关键业务节点手动调用flush()上传增量日志）。

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