如何将Azure Blob中的Databricks DBFS stderr日志发送至同订阅Azure Monitor

以下是几种可行的实现方案，覆盖批量同步和实时推送场景：

方案1：Azure Data Factory/Synapse Pipeline 批量同步 #

• 配置Blob存储为数据源：创建指向存储stderr日志的容器/路径的数据集，选择文本格式并按行解析
• 配置Log Analytics为目标：创建Log Analytics数据集，指定自定义日志表名（需以_CL结尾，比如DatabricksStderr_CL）
• 构建复制/数据流活动：将Blob中的日志文件内容按行读取，映射到自定义表的字段（比如message字段存储日志内容）
• 设置触发规则：配置定时触发（如每小时）或事件触发（当Blob有新文件时），实现增量或全量同步

方案2：Azure Function 实时推送新日志 #

• 创建Blob触发的Function：设置触发路径为存储stderr日志的Blob容器，当新文件上传时自动触发
• 读取并拆分日志：在Function代码中读取文件内容，按行拆分为独立日志条目
• 调用Log Analytics Data Collector API：将日志条目以JSON格式发送到目标工作区，需提前配置工作区ID和共享密钥到Function环境变量
• 权限配置：给Function的托管标识分配Monitoring Metrics Publisher角色，确保具备Log Analytics写入权限

Python示例代码片段：

import azure.functions as func
import requests
import json
import os
import hmac
import hashlib
import base64
from datetime import datetime

WORKSPACE_ID = os.environ["WORKSPACE_ID"]
SHARED_KEY = os.environ["SHARED_KEY"]
LOG_TYPE = "DatabricksStderr"

def generate_signature(date, content_length, method, content_type, resource):
    x_headers = f"x-ms-date:{date}"
    string_to_hash = f"{method}\n{content_length}\n{content_type}\n{x_headers}\n{resource}"
    bytes_to_hash = bytes(string_to_hash, encoding="utf-8")
    decoded_key = base64.b64decode(SHARED_KEY)
    encoded_hash = base64.b64encode(hmac.new(decoded_key, bytes_to_hash, digestmod=hashlib.sha256).digest()).decode()
    return f"SharedKey {WORKSPACE_ID}:{encoded_hash}"

def main(myblob: func.InputStream):
    log_content = myblob.read().decode('utf-8')
    # 过滤空行，构造日志条目
    log_entries = [{"message": line.strip(), "fileName": myblob.name} for line in log_content.split('\n') if line.strip()]
    if not log_entries:
        return
    
    date = datetime.utcnow().strftime("%a, %d %b %Y %H:%M:%S GMT")
    content_length = len(json.dumps(log_entries))
    signature = generate_signature(
        date, content_length, "POST", "application/json", "/api/logs"
    )
    
    headers = {
        "Content-Type": "application/json",
        "Log-Type": LOG_TYPE,
        "x-ms-date": date,
        "Authorization": signature
    }
    
    url = f"https://{WORKSPACE_ID}.ods.opinsights.azure.com/api/logs?api-version=2016-04-01"
    response = requests.post(url, headers=headers, data=json.dumps(log_entries))
    response.raise_for_status()

方案3：Azure Monitor Agent 自定义日志收集（需VM中转） #

• 部署Azure VM：将目标Blob存储通过SMB/NFS协议挂载为VM的本地目录
• 安装Azure Monitor Agent：在VM上部署AMA，关联到目标Log Analytics工作区
• 配置自定义日志规则：创建日志收集规则，指定挂载目录下的stderr日志文件路径，设置按行收集并发送到自定义表

注意事项 #

• 自定义日志表命名必须以_CL结尾，Log Analytics会自动识别并创建对应表结构
• 若日志为结构化格式（如JSON），可在导入阶段配置字段映射，或后续用KQL查询解析
• 确保操作主体（托管标识/服务主体）具备Blob存储的Storage Blob Data Reader权限，以及Log Analytics的Log Analytics Contributor或Monitoring Metrics Publisher权限

内容的提问来源于stack exchange，提问作者K.S Antony Micheal