用Kinesis分析法构建实时会话

Question

2）第二个例子是一个流，它在会话结束后（即5分钟后没有来自用户的数据）为每个会话输出一个事件。这个事件有userId、start_timestamp、end_timestamp和event_count

这在运动分析中是可能的吗？

下面是一个使用Apache Spark进行此操作的示例:https://docs.cloud.databricks.com/docs/latest/databricks_guide/07%20spark%20streaming/applications/01%20sessionization.html

Answer 1

您可以通过创建以下逻辑使用Drools来完成此操作：

类型：

package com.test;

import java.util.List;

declare EventA
    @role( event )
    userId:String;
    seen:boolean;
end

declare SessionStart
    userId: String;
    timestamp: long;
    events: List;
end

declare SessionEnd
    userId: String;
    timestamp: long;
    numOfEvents: int;
end

declare SessionNotification
    userId: String;
    currentNumOfEvents: int;
end

规则：

package com.test;

import java.util.List;
import java.util.ArrayList;

rule "Start session"
when
    // for any EventA
    $a : EventA() from entry-point events
    // check session is not started for this userId
    not (exists(SessionStart(userId == $a.userId)))
then
    modify($a){setSeen(true);}
    List events = new ArrayList();
    events.add($a);
    insert(new SessionStart($a.getUserId(), System.currentTimeMillis(), events));
end

rule "join session"
when
    // for every new EventA
    $a : EventA(seen == false) from entry-point events
    // get event's session
    $session: SessionStart(userId == $a.userId)
then
    $session.getEvents().add($a);
    insertLogical(new SessionNotification($a.getUserId(), $session.getEvents().size()));
    modify($a) {setSeen(true);}

end

rule "End session"
// if session timed out, clean up first
salience 5
when
    // for any EventA
    $a : EventA() from entry-point events
    // check there is no following EventA with same userId within 30 seconds
    not (exists(EventA(this != $a, userId == $a.userId, this after[0, 30s] $a)
        from entry-point events))
    // get event's session
    $session: SessionStart(userId == $a.userId)
then
    insertLogical(new SessionEnd($a.getUserId(), System.currentTimeMillis(),
        $session.getEvents().size()));

    // cleanup
    for (Object $x : $session.getEvents())
        delete($x);
    delete($session);
end