如何存储Python logging模块生成的Azure Function App日志到Blob存储？

下面提供两种可直接落地的实现方案，均会保留日志流的实时可见性，同时将自定义日志写入Blob存储： #

方案1：使用原生Blob输出绑定（代码量最少，无额外依赖） #

该方案利用Azure Functions自带的绑定能力实现，无需额外安装第三方包：

• 第一步：在对应函数的function.json配置文件中添加Blob输出绑定

{
  "name": "outputLogBlob",
  "type": "blob",
  "path": "你的日志容器名/{DateTime:yyyy}/{DateTime:MM}/{DateTime:dd}/func_log_{DateTime:HH}.log",
  "connection": "BLOB_CONN_STR",
  "direction": "out"
}

配置说明：path按时间维度拆分日志路径方便后续检索，connection的值对应你在Function App「配置」页中添加的Blob存储连接字符串的键名。

• 第二步：封装日志处理器实现双端输出

import logging
import azure.functions as func

# 自定义日志缓冲处理器，暂存日志内容
class BlobLogBufferHandler(logging.Handler):
    def __init__(self):
        super().__init__()
        self.buffer = []
    
    def emit(self, record):
        log_line = self.format(record)
        self.buffer.append(log_line)

# 初始化日志实例
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
# 保留默认控制台输出，保证实时日志流可见
console_handler = logging.StreamHandler()
# 初始化Blob日志缓冲处理器
blob_handler = BlobLogBufferHandler()
# 统一日志格式，和日志流显示格式保持一致
log_format = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
console_handler.setFormatter(log_format)
blob_handler.setFormatter(log_format)
logger.addHandler(console_handler)
logger.addHandler(blob_handler)

def main(req: func.HttpRequest, outputLogBlob: func.Out[str]) -> func.HttpResponse:
    # 你的业务逻辑代码
    logger.info("自定义info级日志")
    logger.warning("自定义warning级日志")
    logger.error("自定义error级日志")

    # 函数执行结束前将所有缓冲日志写入Blob
    outputLogBlob.set('\n'.join(blob_handler.buffer))
    return func.HttpResponse("函数执行完成")

方案2：使用Blob SDK实现实时写入（写入延迟最低，适合长运行函数） #

如果你的函数执行时间较长，不想等函数执行完毕才批量写入日志，可以选择该方案：

• 第一步：安装依赖包，将以下内容加入requirements.txt

azure-storage-blob==12.19.0

• 第二步：代码实现

import os
import logging
from datetime import datetime
from azure.storage.blob import BlobServiceClient
import azure.functions as func

# 初始化Blob客户端，连接字符串从Function应用配置中读取
blob_conn_str = os.getenv("BLOB_CONN_STR")
blob_service_client = BlobServiceClient.from_connection_string(blob_conn_str)
log_container_name = "你的日志容器名"

class RealtimeBlobLogHandler(logging.Handler):
    def emit(self, record):
        log_line = self.format(record) + '\n'
        # 按小时生成日志文件名，避免单文件过大
        now = datetime.utcnow()
        blob_path = f"{now.year}/{now.month}/{now.day}/func_log_{now.hour}.log"
        blob_client = blob_service_client.get_blob_client(container=log_container_name, blob=blob_path)
        # 增量追加日志到Blob
        if not blob_client.exists():
            blob_client.create_append_blob()
        blob_client.append_block(log_line)

# 初始化日志实例
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
# 保留控制台输出保证实时日志流可见
console_handler = logging.StreamHandler()
# 初始化实时Blob日志处理器
blob_handler = RealtimeBlobLogHandler()
log_format = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
console_handler.setFormatter(log_format)
blob_handler.setFormatter(log_format)
logger.addHandler(console_handler)
logger.addHandler(blob_handler)

def main(req: func.HttpRequest) -> func.HttpResponse:
    # 你的业务逻辑代码，打日志会实时同步到日志流和Blob
    logger.info("实时写入Blob的自定义日志")
    return func.HttpResponse("函数执行完成")

注意事项 #

• 两种方案均默认保留了控制台日志输出，所有自定义日志都可以在Azure Functions的实时日志流中正常查看
• Blob存储连接字符串统一存储在Function App的应用配置中即可，不要硬编码在业务代码里
• 若日志量较大，可对方案2增加本地批量缓冲逻辑，降低Blob请求次数，减少使用成本

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