如何在Log Analytics中读取ADF输出日志：获取读写行数与处理时长

针对Log Analytics中ADFActivityRun表Output列的非结构化JSON，可通过两种KQL方案提取分散在不同节点的RowsRead和rowsWritten值：

方案1：正则匹配快速提取 #

利用正则表达式全局匹配JSON中所有目标键对应的数值，适合键名固定（注意大小写）的场景：

ADFActivityRun
| where TimeGenerated > ago(7d)  // 按实际需求调整时间范围
| extend OutputDynamic = parse_json(Output)
// 提取所有RowsRead的数值
| extend AllRowsRead = extract_all(@"RowsRead""\s*:\s*""(\d+)""", tostring(OutputDynamic))
// 提取所有rowsWritten的数值
| extend AllRowsWritten = extract_all(@"rowsWritten""\s*:\s*""(\d+)""", tostring(OutputDynamic))
// 按需处理结果：这里以求和为例，也可保留所有原始值
| project 
    PipelineName, 
    ActivityName, 
    RunId,
    AllRowsRead,
    AllRowsWritten,
    TotalRowsRead = array_sum(AllRowsRead),
    TotalRowsWritten = array_sum(AllRowsWritten)

方案2：递归遍历JSON结构（更严谨） #

通过自定义递归函数遍历JSON的所有层级节点，精准提取目标键值，适配任意嵌套深度的结构：

// 定义递归函数：遍历JSON所有节点，提取指定键值
let traverse_json = (json_bag:dynamic) {
    let keys = bag_keys(json_bag);
    union 
        // 递归遍历子节点
        (json_bag
            | mv-apply key = keys on (
                project child_value = json_bag[key]
                | where typeof(child_value) == "dynamic" and isnotnull(child_value)
                | invoke traverse_json(child_value)
            )),
        // 提取当前节点的目标键值
        (json_bag
            | mv-apply key = keys on (
                where key in ("RowsRead", "rowsWritten")
                | project 
                    Key = key, 
                    Value = tolong(json_bag[key])
            ))
};

ADFActivityRun
| where TimeGenerated > ago(7d)
| extend OutputDynamic = parse_json(Output)
// 调用递归函数提取数据
| invoke traverse_json(OutputDynamic)
// 按管道、活动、运行ID聚合求和
| summarize 
    TotalRowsRead = sumif(Value, Key == "RowsRead"),
    TotalRowsWritten = sumif(Value, Key == "rowsWritten")
by PipelineName, ActivityName, RunId

注意事项 #

• 注意键名的大小写：示例中存在RowsRead（首字母大写）和rowsWritten（小写开头），查询中需严格对应；
• 若同一运行存在多个节点的读写数据，可根据需求选择求和、列出所有值或取最大值；
• 确保Output列是合法JSON，若存在格式异常，可添加where isnotnull(Output) and isjson(Output)过滤无效数据。

内容的提问来源于stack exchange，提问作者ADF-Learner