如何有效地分析大量实时数据？

Question

我目前正在开发一个用户网络行为分析工具。 实时用户动作数据达到每秒3000个对象。 我使用python捕获实时数据。你知道吗

我想分析许多用户行为数据：

1. 总共点击了多少个鼠标左/右键？你知道吗
2. 用动作A（B，C，D，E，…）单击了多少鼠标左/右键？你知道吗
3. 在一个小时内，用动作A（B、C、D、E……）点击了多少个鼠标左/右键？你知道吗
4. 同一时间做了多少相同的动作？你知道吗

我想知道如何有效地分析这些实时数据。你知道吗

另外，我正在将这些数据保存在数据库中，但我想知道这些分析是实时的。 另外，我使用了很多If-Else节，这似乎不是最佳代码。 基本上，我在寻找实时分析许多数据的最佳方法。你知道吗

我的主要问题是。。你知道吗

1. 如何实时处理如此多的数据？排队？或者别的什么？你知道吗
2. 如何分析多种类型的分析？我只是用了很多If-Else短语。你知道吗
3. 我可以用numpy、pandas或其他工具来分析这些数据吗？你知道吗

下面的代码是我的一些代码。你知道吗

即将出现的数据对象如下所示：

{
    "usercode": row['usercode'],
    "actiontime": row['actiontime'],
    "actioncode": row['actioncode'],
    "mousebutton": row['mousebutton']
}

数据处理逻辑：

config_action_threshold = 10
config_moment_threshold = 4

memory_collector = dict() 
memory_action = dict()
memory_action_hourly = dict()
memory_moment = dict() 

def addData(item):
    if item['usercode'] not in memory_collector:
        memory_collector[item['usercode']] = {"leftclick":0,"rightclick":0,"actions":0}
        memory_action[item['usercode']] = dict()
        memory_action_hourly[item['usercode']] = {"0006":{},"0612":{},"1218":{},"1824":{}}
        memory_moment[item['usercode']] = {"actiontime":0,"count":0}

    collector = memory_collector[item['usercode']]
    action = memory_action[item['usercode']]
    action_hourly = memory_action_hourly[item['usercode']]
    moment = memory_moment[item['usercode']]

    #normalizing
    if item['actioncode'] not in action:
        action[item['actioncode']] = {"actionL":0,"actionR":0,"actionA":0,"actionB":0}
    actioncode = action[item['actioncode']]

    if item['mousebutton'] == 'L':
        actioncode['L'] += 1
    else:
        actioncode['R'] += 1

    # I have to use if condition because count up when the action occurs over `config_action_threshold`
    if actioncode['L'] >= config_action_threshold:
        if collector['actionL'] == 0:
            collector['actionL'] += config_action_threshold
        else:
            collector['actionL'] += 1

    if actioncode['R'] >= config_action_threshold:
        if collector['actionR'] == 0:
            collector['actionR'] += config_action_threshold
        else:
            collector['actionR'] += 1

    if actioncode['actionA'] >= config_action_threshold:
        if collector['actionA'] == 0:
            collector['actionA'] += config_action_threshold
        else:
            collector['actionA'] += 1

    if actioncode['actionB'] >= config_action_threshold:
        if collector['actionB'] == 0:
            collector['actionB'] += config_action_threshold
        else:
            collector['actionB'] += 1

    if (moment['actiontime'] == item['actiontime']):
        moment['count'] += 1
        if moment['count'] == config_moment_threshold:
            collector['moment'] += config_moment_threshold
        elif moment['count'] > config_moment_threshold:
            collector['moment'] += 1
    else:
        moment['actiontime'] = item['actiontime']
        moment['count'] = 0