如何在Kusto中基于条件计算列的累积和，状态为0时计数重置

需求规则 #

• 按UserId分组计算状态为1的累计天数
• 遇到状态为0时累计值清零，后续再次出现状态1时重新从1开始计数
• 无需校验日期连续性，同用户下只要未遇到状态0就持续累加状态1的计数

输入样例 #

UserId | Day        | Status|
A      |2021-09-16  | 1     |
A      |2021-09-17  | 1     |
A      |2021-09-18  | 1     |
A      |2021-09-19  | 1     |
A      |2021-09-20  | 0     |
A      |2021-09-21  | 1     |
B      |2021-09-16  | 1     |
B      |2021-09-17  | 1     |
B      |2021-09-20  | 1     |
B      |2021-09-21  | 0     |

实现代码 #

SQL版本（支持Hive/Spark SQL/MySQL 8.0及以上版本） #

核心逻辑是用状态0的出现次数给同组的状态1打分组标记，再按分组累加计数：

WITH mark_group AS (
    SELECT 
        UserId,
        Day,
        Status,
        -- 同用户下按日期排序，每遇到一次状态0就给分组标记+1，状态0之后的1会被归为同一组
        SUM(CASE WHEN Status = 0 THEN 1 ELSE 0 END) OVER(PARTITION BY UserId ORDER BY Day) AS group_id
    FROM 你的数据表名
)
SELECT 
    UserId,
    Day,
    Status,
    CASE 
        WHEN Status = 0 THEN 0
        -- 同用户同组内的状态1按日期排序累加计数
        ELSE ROW_NUMBER() OVER(PARTITION BY UserId, group_id, Status ORDER BY Day)
    END AS Cum_Sum
FROM mark_group
ORDER BY UserId, Day;

Python Pandas版本 #

import pandas as pd

# 读入源数据到DataFrame，示例读入代码可根据实际存储调整
# df = pd.read_csv('你的数据文件路径')

# 打分组标记
df['group_id'] = df.groupby('UserId')['Status'].apply(lambda x: (x == 0).cumsum())
# 同组内计数
df['Cum_Sum'] = df.groupby(['UserId', 'group_id', 'Status']).cumcount() + 1
# 状态为0的行计数置为0
df.loc[df['Status'] == 0, 'Cum_Sum'] = 0

输出结果 #

UserId | Day        | Status| Cum_Sum |
A      |2021-09-16  | 1     | 1       |
A      |2021-09-17  | 1     | 2       |
A      |2021-09-18  | 1     | 3       |
A      |2021-09-19  | 1     | 4       |
A      |2021-09-20  | 0     | 0       |
A      |2021-09-21  | 1     | 1       |
B      |2021-09-16  | 1     | 1       |
B      |2021-09-17  | 1     | 2       |
B      |2021-09-20  | 1     | 3       |
B      |2021-09-21  | 0     | 0       |

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