Python 算法实战：基于日志数据统计 N 日活跃用户与 DAU

近期，一个关于“试用期被裁员”的话题引发了广泛讨论。撇开具体的劳动法规，这类事件的背后往往也伴随着绩效评估与数据分析。在工作场景中，业务方（如产品经理或老板）最常提的需求之一便是用户活跃度分析：“最近 N 天的活跃用户有多少？”或“每日活跃用户（DAU）趋势如何？”

这类问题本质上是一个经典的数据处理问题。通常，你只会得到一张最基础的用户行为日志表，核心任务就是从中提取出有价值的业务洞察。今天，我们就用一个具体的算法题——“找出 N 日活跃用户”，来讲解其背后的核心思路，并提供可直接运行的 Python 代码实现。

问题定义：什么是“活跃用户”？

“活跃用户”的定义多样，我们先聚焦于一个最常见且基础的版本：

给定一组用户行为日志，每条记录格式为 (user_id, date)。目标：找出所有在不同日期里，出现次数至少为 k 天的用户（k 为给定的阈值）。

需要注意的两点关键约束：

1. 日期去重：同一天内，同一用户的多次行为仅计为 1 天。
2. 不要求连续：只要用户出现的不同日期总数满足条件即可，日期无需连续。

例如，给定以下示例数据和 k=2：

logs = [
    ("u1", "2024-01-01"),
    ("u1", "2024-01-01"), # 同一天重复，不计为新增天数
    ("u1", "2024-01-02"),
    ("u2", "2024-01-01"),
    ("u2", "2024-01-03"),
    ("u2", "2024-01-04"),
    ("u3", "2024-01-02"),
]
k = 2

用户活跃天数分析：

• u1：出现在 2024-01-01 和 2024-01-02 → 共 2 天
• u2：出现在 2024-01-01、2024-01-03、2024-01-04 → 共 3 天
• u3：出现在 2024-01-02 → 共 1 天

因此，满足 k=2 的活跃用户列表应为：["u1", "u2"]。

核心思路：分组与统计

解决此问题的关键在于“分组统计”。我们可以模拟这个思考过程：

1. 我们需要知道每个 user_id 出现在多少个不同的日期里。
2. 因此，可以为每个用户维护一个集合（Set），用来存储其出现过的所有日期（集合自动去重）。
3. 遍历完所有日志后，检查每个用户的日期集合大小，若其 长度 >= k，则该用户即为目标活跃用户。

这本质上是一个典型的 MapReduce（映射-归约） 思想在单机上的实现：

• Map（分组）：以 user_id 为键，将日期收集到对应的集合中。
• Reduce（筛选）：遍历分组结果，筛选出集合大小满足条件的用户。

Python 代码实现

将上述思路直接翻译成 Python 代码，清晰且高效：

from collections import defaultdict

def find_active_users(logs, k):
    """
    找出在至少 k 个不同日期有行为的用户。
    Args:
        logs: List[Tuple[user_id, date_str]]
        k: 活跃天数阈值
    Returns:
        List[str]: 活跃用户ID列表
    """
    # 1. 分组统计：user_id -> 活跃日期集合
    user_to_days = defaultdict(set)
    for user_id, date_str in logs:
        user_to_days[user_id].add(date_str)  # 利用set自动去重

    # 2. 筛选活跃用户
    active_users = []
    for user_id, days in user_to_days.items():
        if len(days) >= k:
            active_users.append(user_id)

    return active_users

if __name__ == "__main__":
    # 测试数据
    logs = [
        ("u1", "2024-01-01"),
        ("u1", "2024-01-01"),
        ("u1", "2024-01-02"),
        ("u2", "2024-01-01"),
        ("u2", "2024-01-03"),
        ("u2", "2024-01-04"),
        ("u3", "2024-01-02"),
    ]
    k = 2
    print(find_active_users(logs, k))  # 输出: ['u1', 'u2']

代码解析与复杂度分析:

• defaultdict(set)：该数据结构省去了检查键是否存在的逻辑，若键不存在会自动初始化为一个空集合。
• 时间复杂度：O(N + U)。遍历日志列表 O(N)，遍历用户分组结果 O(U)，其中 N 为日志条数，U 为不重复用户数。
• 空间复杂度：O(N)。主要存储 user_to_days，其内容本质是 (user_id, date) 的去重结果。

扩展：如何计算每日活跃用户（DAU）？

业务中另一个常见需求是计算每日活跃用户数（DAU）。这其实是同一个分组统计模型，只是分组键（Key）从“用户”换成了“日期”：

• 原问题：用户 -> {日期1， 日期2...}
• DAU问题：日期 -> {用户1， 用户2...}

实现代码几乎对称：

from collections import defaultdict

def calc_daily_active_users(logs):
    """
    计算每日活跃用户数。
    Args:
        logs: List[Tuple[user_id, date_str]]
    Returns:
        Dict[str, int]: 日期 -> 当日活跃用户数
    """
    # 按日期分组，记录当天活跃的用户集合
    date_to_users = defaultdict(set)
    for user_id, date_str in logs:
        date_to_users[date_str].add(user_id)

    # 计算每日人数
    dau_result = {date_str: len(users) for date_str, users in date_to_users.items()}
    return dau_result

得到 DAU 字典后，便可轻松地用于绘制趋势图、分析活动效果等。

核心模式总结

“活跃用户”分析及其变体（如 DAU、WAU、留存率）是典型的数据处理任务。其核心模式始终是：
选择合适的分组键（用户ID、日期等），然后使用集合（用于去重计数）或计数器进行聚合统计。

掌握这一模式后，面对诸如“最近7天活跃用户”、“每月留存用户”等需求，你的脑海中自然会浮现出 defaultdict(set) 或 defaultdict(int) 的解决方案。通过 Python 进行高效的数据处理，是每一位开发者都需要掌握的核心技能。