SDE2 · Phone Screen Coding

Microsoft SDE2 活跃用户 Top K:无序上下线事件与时间窗裁剪

排序无序上下线事件,求 [start,end) 内各用户在线时长并返回 Top K;说明未闭合会话、并列和非法事件约定。

题目重点:排序无序上下线事件,求 [start,end) 内各用户在线时长并返回 Top K;说明未闭合会话、并列和非法事件约定。

题意与假设

这道练习题保留了面经中值得训练的核心约束,并将需要澄清的条件单独列出。现场拿到变体时,先和面试官确认这些条件。

  • 训练版时间窗为 [start,end),事件时间按整数单位计;合法用户状态在上线/下线之间交替。
  • 同一用户相同时间的多个事件缺少先后语义,代码显式拒绝;同一用户重复上线或未上线却下线也拒绝。
  • 查询结束时仍在线的会话按 end 截断;题目未指定,面试前应确认。
  • 只有出现在 events 中的用户进入候选集;总时长相同时按 userId 升序作训练版稳定并列规则。

解题路线

  1. 先按 timestamp 排序;输入无序,不能直接扫原列表。跨用户同一时间无关,每个用户的重复时间由校验挡下。
  2. 为每个在线用户保存最近上线时刻。读下线事件时,交集时长为 max(0,min(end,offline)−max(start,online)),加到该用户累计值。
  3. 事件扫完后,对仍在线的用户把会话裁剪到 end;再按累计时长降序、userId 升序排序取前 k。
  4. 若数据已按时间流入,可在线累加;若需频繁查询不同时间窗,应维护每用户互不重叠区间及前缀时长索引。

手动走一遍

题目样例无序事件排序后:A [1,4) 贡献 3,B [2,6) 贡献 4,C [3,5) 贡献 2;查询 [1,6)、k=2 得 [B,A]。如果查询改成 [3,5),贡献分别为 A=1、B=2、C=2;训练版同分按 B、C 顺序。

Python 参考实现

下面的代码按上文假设实现,适合本地运行并追加自己的测试用例。

代码区可横向滚动;使用键盘时,先聚焦代码区再按方向键查看。

from collections import defaultdict
from collections.abc import Iterable


def most_active_users(
    events: Iterable[tuple[str, int, bool]], start: int, end: int, k: int
) -> list[str]:
    """Rank [start,end) online durations; each user's times must be distinct."""
    if end < start or k < 0:
        raise ValueError("invalid interval or k")
    ordered = sorted(events, key=lambda event: (event[1], event[0]))
    open_since: dict[str, int] = {}
    duration: dict[str, int] = defaultdict(int)
    seen: set[str] = set()
    last_time: dict[str, int] = {}
    for user, when, online in ordered:
        if last_time.get(user) == when:
            raise ValueError("simultaneous events for one user need a tie rule")
        last_time[user] = when
        seen.add(user)
        if online:
            if user in open_since:
                raise ValueError("duplicate online event")
            open_since[user] = when
        else:
            if user not in open_since:
                raise ValueError("offline without online event")
            begin = open_since.pop(user)
            duration[user] += max(0, min(end, when) - max(start, begin))
    for user, begin in open_since.items():
        duration[user] += max(0, end - max(start, begin))
    return sorted(seen, key=lambda user: (-duration[user], user))[:k]


if __name__ == "__main__":
    sample = [("A", 1, True), ("A", 4, False), ("B", 2, True),
              ("B", 6, False), ("C", 3, True), ("C", 5, False)]
    assert most_active_users(sample, 1, 6, 2) == ["B", "A"]

复杂度

n 个事件排序 O(n log n) 时间;扫描 O(n);至多 u 个用户排序 O(u log u),总 O(n log n+u log u)、空间 O(n+u),其中 n 是事件数、u 是出现过的用户数。只求很小 k 可用大小 k 的堆把末尾排序改为 O(u log k),但前段事件排序仍必要。

易错点与追问

容易失分的边界

  • 按事件输入顺序配对,题目已说完全无序。
  • 把在线时段整段加入而未裁剪查询窗口。
  • 漏掉 end 时仍在线的会话,或擅自假定所有会话已闭合。
  • 把两次上线配成两段重叠时间而重复计时。
  • 把事件数较少当作活跃时长较长。

继续深挖

  • 如果相同时间既下线又上线,如何定义顺序?
  • 若需要支持负时间戳和多时区,事件时间是什么语义?
  • 百万用户、多次区间查询时,如何索引在线区间与 Top K?
  • 是否排除累计 0 的用户及并列分数如何排列?
✓已复制微信号!