python - 在python中按用户计算并发 session

Question

我有一个用户登录和注销表。

表格看起来像这样但是有几十万行:

data = [['aa', '2020-05-31 00:00:01', '2020-05-31 00:00:31'],
        ['bb','2020-05-31 00:01:01', '2020-05-31 00:02:01'],
        ['aa','2020-05-31 00:02:01', '2020-05-31 00:06:03'],
        ['cc','2020-05-31 00:03:01', '2020-05-31 00:04:01'],
        ['dd','2020-05-31 00:04:01', '2020-05-31 00:34:01'],
        ['aa', '2020-05-31 00:05:01', '2020-05-31 00:07:31'],
        ['bb','2020-05-31 00:05:01', '2020-05-31 00:06:01'],
        ['aa','2020-05-31 00:05:01', '2020-05-31 00:08:03'],
        ['cc','2020-05-31 00:10:01', '2020-05-31 00:40:01'],
        ['dd','2020-05-31 00:20:01', '2020-05-31 00:35:01']]


df_test = pd.DataFrame(data,  columns=['user_id','login', 'logout'], dtype='datetime64[ns]')

我能够使用 for 循环以一种 hacky 的方式解决这个问题。它在较小的数据集上运行良好，但在 30 万行上需要数小时。

基本上，这段代码计算每个 session ( session 是每一行)同时登录的用户数

这是我的解决方案。它给出了我需要的结果。我也可以通过使用 apply 编写 lambda 来实现同样的效果，但需要更长的时间。

# create a new column for simultaneous
df_test['simultaneous'] = 0

start_time = time.time()

# loop through dataframe and check condition
for i in df_test.index:
    login, logout = df_test.loc[i,'login'], df_test.loc[i,'logout']
    this_index = df_test.index.isin([i])
    df_test.loc[i, 'simultaneous'] = int(sum(
        (df_test[~this_index]['login'] <= logout) & (df_test[~this_index]['logout'] >= login)
    ))
print("--- %s seconds ---" % (time.time() - start_time))

能否请您看看并告诉我是否有更好的方法来获得相同的结果。也许我遗漏了一些明显的东西。

提前致谢!

Answer 1

此算法基于此数据按登录时间排序这一事实采用流式处理方法。对于每个 session ，它会跟踪注销时间尚未过去的所有 session 的计数(只需将注销时间存储在列表中，并在每次检查新 session 时从该列表中删除陈旧的条目)。我决定将 sess1.logout==sess2.login 计为同时发生，但如果您不同意，可以将 >= 更改为 >。

算法在calculate函数中。

#!/usr/bin/python

import datetime
import random
import time
from statistics import mean, stdev


def calculate(data):
    active_sessions = []
    simultaneous_sessions = []
    for user_id, login, logout in data:
        active_sessions = [ts for ts in active_sessions if ts >= login]
        simultaneous_sessions.append(len(active_sessions))
        active_sessions.append(logout)
    return simultaneous_sessions


def generate_data(numsessions):
    start_time = datetime.datetime(2020, 5, 13, 0, 0, 1)
    data = []
    while len(data) < numsessions:
        for cnt in range(random.choice([0, 0, 0, 1, 1, 2, 3])):
            user_id = chr(ord("a") + cnt) * 2
            duration = random.choice([30, 30, 60, 90, 90, 900, 1800])
            logout_time = start_time + datetime.timedelta(seconds=duration)
            data.append(
                (
                    user_id,
                    start_time.strftime("%Y-%m-%d %H:%M:%S"),
                    logout_time.strftime("%Y-%m-%d %H:%M:%S"),
                )
            )

        start_time += datetime.timedelta(minutes=1)
    return data


start_time = time.time()
num_sessions = 3 * 1e5  # 300,000
print(f"generating data for {num_sessions:.0f} sessions")
data = generate_data(num_sessions)
print(f"sample data=[{data[0]}]")
print("--- %.2f seconds ---" % (time.time() - start_time))
start_time = time.time()
print("calculating simultaneous sessions")
simultaneous_sessions = calculate(data)
print(
    "for {} sessions have max={} min={}, mean={:.2f} stdev={:.2f}".format(
        len(simultaneous_sessions),
        max(simultaneous_sessions),
        min(simultaneous_sessions),
        mean(simultaneous_sessions),
        stdev(simultaneous_sessions),
    )
)
print("--- %.2f seconds ---" % (time.time() - start_time))

从性能的角度来看，我遍历了一次列表，同时不断地重新创建 active_sessions 列表，只要 active_sessions 是一个小数字，就会很快。您可以通过更高效的 active_sessions 列表进行其他优化，但这应该比搜索每个 session 的所有数据快得多。即使数据未按登录时间排序，我认为按登录时间排序然后使用此算法比为每个 session 扫描所有 session 更有效。

更新:我添加了一个合成数据生成器，它基于一些随机变量创建了一堆 session 。这表明该算法处理 30 万行所需的时间不到一秒。

对于 30 万个 session ，大约需要 0.4 秒:

generating data for 300000 sessions
sample data=[('aa', '2020-05-13 00:02:01', '2020-05-13 00:03:31')]
--- 1.99 seconds ---
calculating simultaneous sessions
for 300001 sessions have max=26 min=0, mean=7.42 stdev=2.76
--- 0.35 seconds ---

对于 300 万个 session ，大约需要 4 秒:

generating data for 3000000 sessions
sample data=[('aa', '2020-05-13 00:00:01', '2020-05-13 00:01:31')]
--- 19.35 seconds ---
calculating simultaneous sessions
for 3000001 sessions have max=26 min=0, mean=7.43 stdev=2.77
--- 3.93 seconds ---

O(N)