使用Pandas按用户及连续1分钟时间块分组统计API访问日志

我有包含时间戳（@timestamp）、用户ID（@id）及请求负载的API访问日志，需要按用户分组，把彼此间隔1分钟以内的连续请求归为同一个时间块，统计每个时间块的请求数量。

输入示例 #

@timestamp               @data      @id                                                                                                               
2023-01-21 09:46:33.478   ...      Gh8Z4
2023-01-21 09:46:33.690   ...      Gh8Z4
2023-01-21 09:46:34.189   ...      Gh8Z4
2023-01-21 09:48:28.282   ...      Gh8Z4
2023-01-21 09:51:27.652   ...      HVtpG
2023-01-21 09:51:28.682   ...      Gh8Z4
2023-01-21 09:52:17.412   ...      HVtpG

期望输出示例 #

@id           start                    end             count                                                                                                                  
Gh8Z4 2023-01-21 09:46:33.478  2023-01-21 09:46:34.189   3
Gh8Z4 2023-01-21 09:48:28.282  2023-01-21 09:48:28.282   1  
HVtpG 2023-01-21 09:51:27.652  2023-01-21 09:52:17.412   2
Gh8Z4 2023-01-21 09:51:28.682  2023-01-21 09:51:28.682   1    

解决方案 #

方法1：Python Pandas 处理本地日志 #

如果日志是本地文件或可加载为DataFrame，用Pandas可以快速实现需求：

import pandas as pd

# 加载日志数据（本地文件可改用pd.read_csv/pd.read_table）
data = [
    ["2023-01-21 09:46:33.478", "...", "Gh8Z4"],
    ["2023-01-21 09:46:33.690", "...", "Gh8Z4"],
    ["2023-01-21 09:46:34.189", "...", "Gh8Z4"],
    ["2023-01-21 09:48:28.282", "...", "Gh8Z4"],
    ["2023-01-21 09:51:27.652", "...", "HVtpG"],
    ["2023-01-21 09:51:28.682", "...", "Gh8Z4"],
    ["2023-01-21 09:52:17.412", "...", "HVtpG"],
]

df = pd.DataFrame(data, columns=["@timestamp", "@data", "@id"])
# 解析时间戳为datetime类型
df["@timestamp"] = pd.to_datetime(df["@timestamp"])

result_list = []
# 按用户ID分组处理每个用户的请求
for user_id, group in df.groupby("@id"):
    # 对当前用户的请求按时间排序
    sorted_group = group.sort_values("@timestamp").reset_index(drop=True)
    # 计算相邻请求的时间差（秒）
    sorted_group["time_gap"] = sorted_group["@timestamp"].diff().dt.total_seconds()
    # 标记新时间块的起始点：时间差超60秒或为第一条请求
    sorted_group["block_id"] = (sorted_group["time_gap"] > 60).cumsum()
    # 统计每个时间块的起始、结束时间和请求数
    block_stats = sorted_group.groupby("block_id").agg(
        start=("@timestamp", "first"),
        end=("@timestamp", "last"),
        count=("@timestamp", "size")
    ).reset_index(drop=True)
    block_stats["@id"] = user_id
    result_list.append(block_stats[["@id", "start", "end", "count"]])

# 合并结果并按时间排序
final_result = pd.concat(result_list).sort_values("start")
# 格式化时间输出为指定格式
final_result["start"] = final_result["start"].dt.strftime("%Y-%m-%d %H:%M:%S.%f").str[:-3]
final_result["end"] = final_result["end"].dt.strftime("%Y-%m-%d %H:%M:%S.%f").str[:-3]

# 打印结果
print(final_result.to_string(index=False))

方法2：SQL 处理数据库日志表 #

如果日志存储在数据库中，可通过窗口函数实现分组统计：

WITH sorted_logs AS (
    -- 按用户分组排序，获取上一条请求的时间戳
    SELECT 
        @id,
        @timestamp,
        LAG(@timestamp) OVER (PARTITION BY @id ORDER BY @timestamp) AS prev_time
    FROM api_logs
),
block_flags AS (
    -- 标记新时间块的起始：时间差超60秒或无历史记录
    SELECT 
        @id,
        @timestamp,
        CASE 
            WHEN TIMESTAMPDIFF(SECOND, prev_time, @timestamp) > 60 OR prev_time IS NULL 
            THEN 1 
            ELSE 0 
        END AS is_new_block
    FROM sorted_logs
),
block_groups AS (
    -- 为每个用户的请求分配时间块ID
    SELECT 
        @id,
        @timestamp,
        SUM(is_new_block) OVER (PARTITION BY @id ORDER BY @timestamp) AS block_id
    FROM block_flags
)
-- 统计每个时间块的信息
SELECT 
    @id,
    MIN(@timestamp) AS start,
    MAX(@timestamp) AS end,
    COUNT(*) AS count
FROM block_groups
GROUP BY @id, block_id
ORDER BY start;

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