使用 Kinesis Analytics 构建实时会话答案

【问题标题】：Using Kinesis Analytics to construct real time sessions使用 Kinesis Analytics 构建实时会话
【发布时间】：2018-09-19 22:19:25
【问题描述】：

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

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

通常这是在 SQL 中使用 LAG 函数完成的，因此您可以计算连续行之间的时间差。这篇文章：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

但我很想用一个（或两个）Kinesis Analytics 流来做到这一点。

【问题讨论】：

标签： amazon-kinesis bigdata

【解决方案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

您可以使用this service 编写 Drools Kinesis Analytics

【讨论】：

【解决方案2】：

Kinesis Analytics 现在支持LAG。您可以在文档页面http://docs.aws.amazon.com/kinesisanalytics/latest/sqlref/sql-reference-lag.html 上看到它。我实际上已经将它用于与您描述的类似的用例。

【讨论】：