如何使用KQL实现未收到更新时将数据结转至下一小时

核心逻辑 #

要实现当前小时未收到产品上报时自动结转上一小时指标值，按以下步骤处理即可：

• 提取数据集中所有出现过的小时时间点、所有产品，生成「时间-产品」的全量笛卡尔组合，保证每个小时维度下所有产品都有对应行
• 将全量组合与原始上报数据做左连接，匹配到原始上报的行取实际上报值，未匹配到的数值位留空
• 按产品分组、按时间升序排序，用前向填充逻辑取上一个非空值补全空位，过滤掉产品首次上报前无历史值的无效行即可，该逻辑同时支持连续多小时缺数的自动结转

完整实现代码 #

以下代码包含样例数据构造，可直接运行验证：

// 样例输入数据集
let RawData = datatable(Time:datetime, Product:string, Value:long)
[
    datetime(2022-06-20T01:00:00Z), "B1", 25,
    datetime(2022-06-20T01:00:00Z), "B2", 46,
    datetime(2022-06-20T01:00:00Z), "B3", 18,
    datetime(2022-06-20T02:00:00Z), "B1", 26,
    datetime(2022-06-20T02:00:00Z), "B2", 40,
];
// 提取全量时间、全量产品维度
let AllTimeSlots = RawData | distinct Time;
let AllProducts = RawData | distinct Product;
AllTimeSlots
| extend TempJoinKey = 1
| join kind=inner (AllProducts | extend TempJoinKey = 1) on TempJoinKey
| project Time, Product
// 左连原始数据，缺失上报的行Value为空
| join kind=leftouter RawData on Time, Product
| project Time, Product, Value = coalesce(Value, long(null))
// 按产品分组做前向填充
| partition by Product
(
    order by Time asc
    | serialize PrevValue = prev(Value, 1, long(null))
    | extend Value = coalesce(Value, PrevValue)
)
// 过滤无历史值可结转的空行
| where isnotnull(Value)
| sort by Time asc, Product asc
| project Time, Product, Value

运行结果 #

执行代码后输出结果如下，2022-06-20T02:00:00时点缺失上报的B3产品，会自动结转上一小时的18作为当前小时指标值：

Time                     Product Value
2022-06-20T01:00:00Z     B1      25
2022-06-20T01:00:00Z     B2      46
2022-06-20T01:00:00Z     B3      18
2022-06-20T02:00:00Z     B1      26
2022-06-20T02:00:00Z     B2      40
2022-06-20T02:00:00Z     B3      18

注：你提供的预期输出中B3结转记录的时间标注为01:00应为排版笔误，结转逻辑是将历史值填充到当前小时的时间槽位，符合「结转至当前小时」的需求。

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