使用 Kinesis Analytics 构建实时会话

是否有某个地方的示例，或者有人可以解释如何使用 Kinesis Analytics 构建实时会话。 （即会话化）

这里提到这可能：https://aws.amazon.com/blogs/aws/amazon-kinesis-analytics-process-streaming-data-in-real-time-with-sql/ https://aws.amazon.com/blogs/aws/amazon-kinesis-analytics-process-streaming-data-in-real-time-with-sql/在自定义窗口的讨论中但没有给出示例。

通常，这是使用 LAG 函数在 SQL 中完成的，因此您可以计算连续行之间的时间差。这个帖子：https://blog.modeanalytics.com/finding-user-sessions-sql/ https://blog.modeanalytics.com/finding-user-sessions-sql/描述如何使用传统（非流式）SQL 来完成此操作。但是，我在 Kinesis Analytics 中没有看到对 LAG 功能的支持。

我特别喜欢两个例子。假设两者都采用由 user_id 和时间戳组成的流作为输入。将会话定义为来自同一用户、间隔不超过 5 分钟的一系列事件

1) 第一个输出具有附加列 event_count session_start_timestamp 的流。每次出现事件时，都应该输出带有这两个附加列的事件。

2) 第二个示例是一个流，一旦会话结束（即 5 分钟过去，没有来自用户的数据），每个会话输出一个事件。该事件将包含 userId、start_timestamp、end_timestamp 和 event_count

Kinesis Analytics 可以实现这一点吗？

以下是使用 Apache Spark 执行此操作的示例：https://docs.cloud.databricks.com/docs/latest/databricks_guide/07%20Spark%20Streaming/Applications/01%20Sessionization.html https://docs.cloud.databricks.com/docs/latest/databricks_guide/07%20Spark%20Streaming/Applications/01%20Sessionization.html

但我很乐意使用一个（或两个）Kinesis Analytics 流来完成此操作。

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

Types:

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

Rules:

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

您可以使用 Drools Kinesis Analytics 创作这项服务 https://aws.amazon.com/marketplace/pp/B075MDR5DC