Azure Data Explorer更新策略中无效Payload的处理与监控方法问询

一、分流存储不符合验证的数据 #

你可以通过双Update Policy的方式，将符合验证的数据导入目标业务表，不符合的导入专门的错误表（如RawData_Invalid），同时记录具体的错误原因，方便后续排查。

1. 封装可复用的验证逻辑（推荐）

创建Kusto函数统一处理验证规则，避免重复代码：

.create function ValidateSensorPayload(RawPayload: dynamic, TargetLocation: string)
{
    let SensorId = tostring(RawPayload['Sensor-ID']);
    let PayloadVersion = tostring(RawPayload['PayloadStructureVersion']);
    let SensorLocation = tostring(RawPayload['Sensor-Location']);
    let PayloadUnit = tostring(RawPayload['Unit']);

    // 逐步验证并收集错误信息
    let ValidationErrors = dynamic([])
        // 验证Sensor-ID格式
        | extend Errors = iff(SensorId matches regex "[0-9]{3}-[A-Z]{3}", Errors, array_append(Errors, "Invalid Sensor-ID format"))
        // 验证Payload版本
        | extend Errors = iff(PayloadVersion in ("v2","v3"), Errors, array_append(Errors, "Unsupported PayloadStructureVersion"))
        // 验证目标位置匹配
        | extend Errors = iff(SensorLocation == TargetLocation, Errors, array_append(Errors, "Mismatched Sensor-Location for target table"))
        // 验证Unit与传感器配置匹配
        | join kind=leftouter (Sensors) on $left.SensorId == $right.SensorId, $left.PayloadUnit == $right.Unit
        | extend Errors = iff(isnotempty(Unit), Errors, array_append(Errors, "Unit does not match sensor configuration"))
        | project Errors;

    let IsValid = array_length(ValidationErrors) == 0;
    // 返回验证结果和关键字段
    pack('IsValid', IsValid, 'ValidationErrors', ValidationErrors, 'SensorId', SensorId, 'PayloadVersion', PayloadVersion, 'SensorLocation', SensorLocation, 'PayloadUnit', PayloadUnit)
}

2. 配置两个Update Policy

• Policy 1：导入有效数据到业务表（如Station1_Data）

RawData
| extend ValidationResult = ValidateSensorPayload(['payload'], "Station1")
| evaluate bag_unpack(ValidationResult)
| where IsValid == true
// 提取业务所需字段
| project 
    SensorId, 
    PayloadStructureVersion = PayloadVersion, 
    SensorLocation, 
    Value = todouble(['payload']['Value']), 
    Unit = PayloadUnit, 
    Timestamp = todatetime(['payload']['Timestamp'])

• Policy 2：导入无效数据到错误表（如RawData_Invalid）

RawData
| extend ValidationResult = ValidateSensorPayload(['payload'], "Station1")
| evaluate bag_unpack(ValidationResult)
| where IsValid == false
// 保留原始Payload、错误信息和摄入时间
| project 
    RawPayload = ['payload'], 
    ValidationErrors, 
    IngestionTimestamp = ingestion_time()

二、监控不符合验证的数据 #

1. 实时错误统计查询

直接查询错误表，分析错误类型、时间分布：

RawData_Invalid
// 将错误数组转为逗号分隔的字符串，便于分组统计
| extend ErrorDetails = array_join(ValidationErrors, ", ")
| summarize 
    ErrorCount = count() 
    by bin(IngestionTimestamp, 1h), ErrorDetails
// 可视化展示
| render columnchart

2. 设置告警规则

通过Azure Monitor配置告警，当错误量超过阈值时触发通知：

• 选择ADX集群作为数据源，设置查询规则为RawData_Invalid | summarize count() by bin(ingestion_time(), 5m)
• 设置阈值（比如5分钟内错误数≥100），配置通知方式（邮件、Teams等）

3. 配置数据保留策略

为错误表设置合理的保留期，避免占用过多存储：

.alter table RawData_Invalid policy retention softdelete = 30d

核心最佳实践 #

• 明确分流：有效数据和错误数据物理隔离，不影响业务表的纯净性
• 精准错误标记：记录具体错误原因，而非仅标记"无效"，降低排查成本
• 逻辑复用：封装验证函数，多表场景下统一维护规则
• 主动监控：结合ADX查询能力和Azure Monitor，实现错误的实时感知

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