如何在Azure Durable Functions中记录机器学习训练步骤？

结论：推荐将训练日志（包括epoch日志）存储到Azure Blob Storage #

• Azure Functions默认的Application Insights日志偏向函数执行监控，不适合结构化训练日志的持久化与后续分析
• Blob Storage提供低成本、高可靠的持久化存储，支持日志的长期留存、模型溯源与离线分析
• 可直接与PyTorch Lightning的日志系统集成，无需复杂改造

具体实现方法与代码 #

前提准备

1. 在Azure Functions应用设置中添加AZURE_STORAGE_CONNECTION_STRING，值为你的Blob存储账户连接字符串
2. 确保目标Blob容器（如training-logs）已创建，或在代码中添加自动创建逻辑
3. 安装依赖包：

pip install azure-storage-blob pytorch-lightning azure-functions-durable

方法1：训练完成后上传CSVLogger生成的日志（简单易用）

利用PyTorch Lightning原生CSVLogger生成本地日志，训练结束后批量上传到Blob Storage，适合大多数场景。

import os
import shutil
from pytorch_lightning import LightningModule, Trainer
from pytorch_lightning.loggers import CSVLogger
from azure.storage.blob import BlobServiceClient

# 你的PyTorch Lightning模型示例
class MyMLModel(LightningModule):
    def training_step(self, batch, batch_idx):
        loss = ...  # 替换为你的损失计算逻辑
        # 记录单步训练损失 + epoch平均损失
        self.log('train_loss_step', loss, on_step=True, logger=True)
        self.log('train_loss_epoch', loss, on_epoch=True, logger=True)
        return loss
    
    def validation_step(self, batch, batch_idx):
        val_loss = ...  # 替换为你的验证损失计算逻辑
        # 记录验证集epoch损失
        self.log('val_loss_epoch', val_loss, on_epoch=True, logger=True)

def train_and_upload_logs(context):
    # 初始化CSVLogger，临时存储到函数本地目录
    logger = CSVLogger(save_dir='./temp_training_logs', name='image_classification')
    
    # 初始化训练器
    trainer = Trainer(
        max_epochs=15,
        logger=logger,
        enable_checkpointing=False  # 根据需求开启
    )
    
    # 执行训练
    model = MyMLModel()
    trainer.fit(
        model,
        train_dataloaders=your_train_dataloader,
        val_dataloaders=your_val_dataloader
    )
    
    # 上传日志到Blob Storage
    blob_service_client = BlobServiceClient.from_connection_string(
        os.getenv("AZURE_STORAGE_CONNECTION_STRING")
    )
    container_name = "training-logs"
    log_file_path = os.path.join(logger.log_dir, 'metrics.csv')
    
    # 上传文件到指定Blob路径
    blob_client = blob_service_client.get_blob_client(
        container=container_name,
        blob=f"{logger.name}/{logger.version}/training_metrics.csv"
    )
    with open(log_file_path, "rb") as data:
        blob_client.upload_blob(data, overwrite=True)
    
    # 清理本地临时文件
    shutil.rmtree('./temp_training_logs')
    
    return f"训练完成，日志已上传至Blob: {container_name}/{logger.name}/{logger.version}/training_metrics.csv"

方法2：自定义Logger实时写入Blob（适合实时监控）

如果需要实时记录epoch日志（无需等待训练结束），可以自定义PyTorch Lightning Logger，直接将指标写入Blob Storage。

from pytorch_lightning.loggers.logger import Logger
from pytorch_lightning.utilities.rank_zero import rank_zero_only
from azure.storage.blob import BlobServiceClient
import csv
import os
from io import StringIO
from datetime import datetime

class AzureBlobCSVLogger(Logger):
    def __init__(self, connection_string, container_name, experiment_name):
        super().__init__()
        self.connection_string = connection_string
        self.container_name = container_name
        self.experiment_name = experiment_name
        self.blob_service_client = BlobServiceClient.from_connection_string(connection_string)
        self.metrics_buffer = []
        self.fieldnames = None
        # 初始化容器（如果不存在）
        self._create_container_if_not_exists()

    def _create_container_if_not_exists(self):
        if not self.blob_service_client.get_container_client(self.container_name).exists():
            self.blob_service_client.create_container(self.container_name)

    @property
    def name(self):
        return self.experiment_name

    @property
    def version(self):
        # 用时间戳作为版本号，避免重复
        return datetime.now().strftime("%Y%m%d_%H%M%S")

    @rank_zero_only
    def log_metrics(self, metrics, step=None):
        # 初始化表头（第一次记录时）
        if self.fieldnames is None:
            self.fieldnames = list(metrics.keys()) + ['step'] if step else list(metrics.keys())
        
        # 追加当前指标到缓冲区
        metrics['step'] = step
        self.metrics_buffer.append(metrics)
        
        # 每epoch写入一次（可根据需求调整为每N步写入）
        if 'val_loss_epoch' in metrics:
            self._write_to_blob()

    @rank_zero_only
    def _write_to_blob(self):
        output = StringIO()
        writer = csv.DictWriter(output, fieldnames=self.fieldnames)
        
        # 第一次写入表头
        if len(self.metrics_buffer) == len([m for m in self.metrics_buffer if 'val_loss_epoch' in m]):
            writer.writeheader()
        
        writer.writerows(self.metrics_buffer)
        
        # 上传到Blob
        blob_client = self.blob_service_client.get_blob_client(
            container=self.container_name,
            blob=f"{self.name}/{self.version}/real_time_metrics.csv"
        )
        blob_client.upload_blob(output.getvalue(), overwrite=True)
        
        # 清空缓冲区
        self.metrics_buffer = []

    @rank_zero_only
    def save(self):
        # 训练结束时确保剩余日志写入
        if self.metrics_buffer:
            self._write_to_blob()

    @rank_zero_only
    def finalize(self, status):
        self.save()

# 在Durable Functions中使用自定义Logger
def train_with_real_time_logs(context):
    logger = AzureBlobCSVLogger(
        connection_string=os.getenv("AZURE_STORAGE_CONNECTION_STRING"),
        container_name="training-logs",
        experiment_name="real_time_training"
    )
    
    trainer = Trainer(
        max_epochs=15,
        logger=logger
    )
    
    model = MyMLModel()
    trainer.fit(
        model,
        train_dataloaders=your_train_dataloader,
        val_dataloaders=your_val_dataloader
    )
    
    return "训练完成，实时epoch日志已保存至Blob Storage"

关键说明 #

• epoch日志捕获：通过在training_step和validation_step中使用self.log(..., on_epoch=True)，可以自动记录epoch级别的平均指标，PyTorch Lightning会在epoch结束时将这些指标传递给Logger
• 权限配置：确保Azure Functions的系统分配身份或连接字符串拥有Blob Storage的Storage Blob Data Contributor权限
• 临时目录：Azure Functions的本地临时目录（./temp_*）有写入权限，但重启后会被清空，因此训练后需及时上传日志

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