如何通过SQL或Power Query实现工作流时序状态准确统计

场景说明 #

• 业务基于Web表单搭建工作流，共设4种状态：1.Started（流程发起）、2.Review by Admin（管理员审核）、3.Back to User（退回用户修改）、4.Finished（流程办结）
• 状态2、3支持按业务规则循环流转，全量操作日志存入MS SQL数据库，需统计工作流随时间的演变情况
• 统计规则约束：
  • 状态2、3循环产生的重复状态记录不得重复计数
  • 流程首次发起后，重复产生的状态1记录不纳入统计
  • 结果支持长表格式，计数准确即可
• 已尝试Power Query分组、dense_rank等方案未达预期，提供MS SQL、Power Query两种实现路径如下

核心实现逻辑 #

所有方案统一遵循两层去重逻辑，从根源避免重复计数：

1. 相邻去重：单工作流日志按操作时间升序排列后，仅保留与上一条记录状态不同的条目，过滤异常插入的连续重复状态（含重复的状态1）
2. 首次进入去重：单工作流维度下，每个状态仅保留第一次进入的时间记录，自动过滤2、3循环往返产生的重复状态条目

方案1：MS SQL 实现 #

假设原始日志表名为workflow_logs，核心字段如下，可根据实际业务字段替换：

• workflow_id：工作流唯一ID
• status_id：状态ID（1/2/3/4）
• status_name：状态名称
• operate_time：操作时间

完整实现代码：

WITH sorted_logs AS (
    SELECT
        *,
        -- 取同工作流上一条操作的状态，用于相邻重复判断
        LAG(status_id) OVER (
            PARTITION BY workflow_id 
            ORDER BY operate_time
        ) AS prev_status
    FROM workflow_logs
),
adj_deduped_logs AS (
    -- 第一层：相邻重复状态去重
    SELECT
        workflow_id,
        status_id,
        status_name,
        operate_time
    FROM sorted_logs
    WHERE
        prev_status IS NULL -- 保留流程第一条发起记录
        OR status_id != prev_status -- 仅保留状态发生变化的记录
),
first_enter_logs AS (
    -- 第二层：取每个工作流各状态的首次进入时间，过滤循环重复
    SELECT
        workflow_id,
        status_id,
        status_name,
        MIN(operate_time) AS first_enter_time
    FROM adj_deduped_logs
    GROUP BY workflow_id, status_id, status_name
)
-- 按日统计各状态进入流程量示例，可按需调整时间维度
SELECT
    CONVERT(DATE, first_enter_time) AS stat_date,
    status_id,
    status_name,
    COUNT(workflow_id) AS workflow_count
FROM first_enter_logs
GROUP BY CONVERT(DATE, first_enter_time), status_id, status_name
ORDER BY stat_date, status_id

如果需要保留全量状态流转节点（仅过滤连续重复、不合并循环的2、3节点），直接使用adj_deduped_logs层结果做统计即可。

方案2：Power Query 实现 #

以下M代码可直接在Power Query高级编辑器中修改使用，默认从MS SQL源读取日志数据：

let
    // 连接MS SQL读取原始日志，可替换为本地Excel/其他数据源
    Source = Sql.Database("你的SQL实例地址", "你的业务库名", [Query="SELECT workflow_id, status_id, status_name, operate_time FROM workflow_logs"]),
    // 按工作流ID分组处理
    GroupByWorkflow = Table.Group(Source, {"workflow_id"}, {
        {"WorkflowProcess", (processTable) =>
            let
                // 组内按操作时间升序排序
                Sorted = Table.Sort(processTable,{{"operate_time", Order.Ascending}}),
                // 添加上一行状态列做相邻去重判断
                AddPrevStatus = Table.AddColumn(Sorted, "prev_status", (row, rowIndex) => 
                    if rowIndex = 0 then null else Sorted[status_id]{rowIndex-1}, 
                    Int64.Type
                ),
                // 过滤相邻重复状态
                AdjDedup = Table.SelectRows(AddPrevStatus, each [prev_status] = null or [status_id] <> [prev_status]),
                // 按状态分组取首次进入时间，过滤循环重复
                FirstEnter = Table.Group(AdjDedup, {"status_id", "status_name"}, {
                    {"first_enter_time", each List.Min([operate_time]), type datetime}
                })
            in
                FirstEnter
        , type table [status_id=Int64.Type, status_name=text, first_enter_time=datetime]}
    }),
    // 展开分组结果
    ExpandResult = Table.ExpandTableColumn(GroupByWorkflow, "WorkflowProcess", {"status_id", "status_name", "first_enter_time"}, {"status_id", "status_name", "first_enter_time"}),
    // 以下为按日统计示例，可按需调整统计维度
    AddStatDate = Table.AddColumn(ExpandResult, "stat_date", each DateTime.Date([first_enter_time]), type date),
    FinalStat = Table.Group(AddStatDate, {"stat_date", "status_id", "status_name"}, {{"workflow_count", each Table.RowCount(_), Int64.Type}})
in
    FinalStat

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