如何在Linux服务器通过Logs Ingestion API每日导入Pandas数据至Log Analytics Workspace？

实现步骤 #

Azure端准备 #

• 创建Data Collection Endpoint（DCE）和Data Collection Rule（DCR）：
  • 在Azure门户创建DCE，记录其端点URL（格式类似https://<dce-name>.<region>.ingest.monitor.azure.com）
  • 创建DCR并关联目标Log Analytics Workspace，配置自定义表的字段映射（需与你的DataFrame字段匹配），记录DCR的Immutable ID和目标表名（格式为Custom_<TableName>_CL）
• 创建Azure AD服务主体：
  • 在Azure AD中注册应用，获取客户端ID、客户端密钥、租户ID
  • 给该服务主体分配Monitoring Data Publisher角色到DCR资源，确保具备数据写入权限

Python脚本实现（Linux环境） #

• 安装依赖包：

  pip install pandas requests azure-identity
  

• 编写核心逻辑：
  1. 生成或加载目标Pandas DataFrame
  2. 将DataFrame转换为API要求的JSON结构（列表嵌套字典）
  3. 通过Azure Identity获取访问令牌
  4. 调用Logs Ingestion API发送数据
• 测试脚本：运行后检查Log Analytics自定义表是否成功接收数据

Linux定时任务配置 #

• 编辑crontab：

  crontab -e
  

• 添加每日执行任务（示例为凌晨1点运行）：

  0 1 * * * /usr/bin/python3 /path/to/your/script.py >> /path/to/run_log.log 2>&1
  

  注意替换Python路径、脚本路径和日志路径，确保脚本拥有执行权限

最小可复现代码示例 #

import pandas as pd
import requests
from azure.identity import ClientSecretCredential

# 替换为你的实际配置参数
TENANT_ID = "your-tenant-id"
CLIENT_ID = "your-client-id"
CLIENT_SECRET = "your-client-secret"
DCE_ENDPOINT = "https://<dce-name>.<region>.ingest.monitor.azure.com"
DCR_ID = "/subscriptions/<subscription-id>/resourceGroups/<rg-name>/providers/Microsoft.Insights/dataCollectionRules/<dcr-name>"
TABLE_NAME = "Custom_YourTableName_CL"

def get_ingestion_token():
    """获取Logs Ingestion API的访问令牌"""
    credential = ClientSecretCredential(
        tenant_id=TENANT_ID,
        client_id=CLIENT_ID,
        client_secret=CLIENT_SECRET
    )
    token = credential.get_token("https://monitor.azure.com/.default")
    return token.token

def send_data_to_log_analytics(df):
    """将DataFrame数据发送到Log Analytics"""
    token = get_ingestion_token()
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    # 转换DataFrame为API兼容格式
    payload = df.to_dict("records")
    # 构建请求URL，注意数据流名称格式为Custom-<表名前缀>
    stream_name = f"Custom-{TABLE_NAME.split('_')[1]}"
    url = f"{DCE_ENDPOINT}/dataCollectionRules/{DCR_ID.split('/')[-1]}/streams/{stream_name}?api-version=2021-11-01-preview"
    
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    print(f"数据发送成功，响应状态码: {response.status_code}")

if __name__ == "__main__":
    # 替换为你的实际数据生成逻辑
    test_data = {
        "Timestamp": pd.date_range(start="2024-01-01", periods=5, freq="H"),
        "MetricValue": [15, 25, 35, 45, 55],
        "Status": ["OK", "OK", "Warning", "OK", "Critical"]
    }
    df = pd.DataFrame(test_data)
    
    try:
        send_data_to_log_analytics(df)
    except Exception as e:
        print(f"数据发送失败: {str(e)}")

注意事项 #

• 确保DCR配置的数据流名称与脚本中stream_name一致
• 时间戳字段需符合ISO 8601格式，Log Analytics会自动识别Timestamp字段为时间索引
• 可根据需求扩展脚本的日志记录和错误重试逻辑

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