【问题标题】：Creating different sessions based on unique key根据唯一键创建不同的会话
【发布时间】：2019-04-09 05:30:58
【问题描述】：

我从一个向我发送 JSON 消息的 kafka 主题获取消息。我想从该 json 消息中提取一个字段（可以是一个 ID），并且我想为“n”个唯一设备 ID 创建“n”个会话。

我已尝试为收到的每个唯一 ID 创建一个新会话实例，但在创建新会话窗口实例后，即在管道中为每个 ID 创建一个新分支后，我无法将下一个即将到来的消息推送到对应的已经存在的分支。

我想要的预期结果是，假设我们收到这样的消息

{ID:1,...}, {ID:2,...}, {ID:3,...},{ID:1,...}

将创建三个不同的会话，第四条消息将发送到设备 ID 1 的会话。有没有办法在 apache 梁编程范式或 Java 编程范式中做到这一点？任何帮助将不胜感激。

【问题讨论】：

会话是什么意思？您是否想将所有消息分组到单个设备并在例如设备上运行？他们的名单？你看过会话窗口吗？这张地图是否更符合您的要求？
这是我试图实现的一种用例。我已经使用了会话窗口，但我的要求是每个 ID 应该有一个不同的会话，具有不同的间隙持续时间，在默认会话中，它会为每个键创建会话，但我们不能为每个键分配不同的间隙持续时间。

标签： java google-cloud-dataflow apache-beam

【解决方案1】：

是的，如果您使用自定义 WindowFn，则可以使用 Beam 范例。您可以继承 Sessions 类并对其进行修改，以根据每个元素的 ID 设置不同的间隔持续时间。您可以在assignWindows 中执行此操作，在Sessions 中看起来像这样：

  @Override
  public Collection<IntervalWindow> assignWindows(AssignContext c) {
    // Assign each element into a window from its timestamp until gapDuration in the
    // future.  Overlapping windows (representing elements within gapDuration of
    // each other) will be merged.
    return Arrays.asList(new IntervalWindow(c.timestamp(), gapDuration));
  }

AssignContext 类可用于访问分配给此窗口的元素，这将允许您检索该元素的 ID。

听起来您还希望将具有不同 ID 的元素分组在不同的窗口中（即，如果元素 A 和 B 在间隙持续时间内进入但具有不同的 ID，则它们仍应位于不同的窗口中）。这可以通过使用元素的 ID 作为键执行 GroupByKey 来完成。会话窗口适用于每个键as described in the Beam Programming Guide，因此这将按 ID 分隔元素。

【讨论】：

【解决方案2】：

我已经为此用例实现了 Java 和 Python 示例。 Java 遵循 Daniel Oliveira 建议的方法，但我认为分享一个工作示例很好。

请注意，Java 示例在 Beam 常见模式 docs 中有所体现。 Python（使用 fnapi）不支持自定义合并窗口。

Java 版本：

我们可以调整来自Session windows 的代码以适应我们的用例。

简而言之，当记录被窗口化为会话时，它们被分配到一个从元素的时间戳开始的窗口（未对齐的窗口），并将间隙持续时间添加到开始以计算结束。然后mergeWindows 函数将组合每个键的所有重叠窗口，从而延长会话。

我们需要修改 assignWindows 函数以创建一个具有数据驱动间隙而不是固定持续时间的窗口。我们可以通过WindowFn.AssignContext.element() 访问该元素。原来的赋值函数是：

public Collection<IntervalWindow> assignWindows(AssignContext c) {
  // Assign each element into a window from its timestamp until gapDuration in the
  // future.  Overlapping windows (representing elements within gapDuration of
  // each other) will be merged.
  return Arrays.asList(new IntervalWindow(c.timestamp(), gapDuration));
}

修改后的函数为：

@Override
public Collection<IntervalWindow> assignWindows(AssignContext c) {
  // Assign each element into a window from its timestamp until gapDuration in the
  // future.  Overlapping windows (representing elements within gapDuration of
  // each other) will be merged.
  Duration dataDrivenGap;
  JSONObject message = new JSONObject(c.element().toString());

  try {
    dataDrivenGap = Duration.standardSeconds(Long.parseLong(message.getString(gapAttribute)));
  }
  catch(Exception e) {
    dataDrivenGap = gapDuration;
  }
  return Arrays.asList(new IntervalWindow(c.timestamp(), dataDrivenGap));
}

请注意，我们添加了一些额外的东西：

默认值用于数据中不存在自定义间隙的情况
一种将主管道中的属性设置为自定义窗口的方法。

withDefaultGapDuration 和 withGapAttribute 方法是：

/** Creates a {@code DynamicSessions} {@link WindowFn} with the specified gap duration. */
public static DynamicSessions withDefaultGapDuration(Duration gapDuration) {
  return new DynamicSessions(gapDuration, "");
}

public DynamicSessions withGapAttribute(String gapAttribute) {
  return new DynamicSessions(gapDuration, gapAttribute);
}

我们还将添加一个新字段 (gapAttribute) 和构造函数：

public class DynamicSessions extends WindowFn<Object, IntervalWindow> {
  /** Duration of the gaps between sessions. */
  private final Duration gapDuration;

    /** Pub/Sub attribute that modifies session gap. */
  private final String gapAttribute;

  /** Creates a {@code DynamicSessions} {@link WindowFn} with the specified gap duration. */
  private DynamicSessions(Duration gapDuration, String gapAttribute) {
    this.gapDuration = gapDuration;
    this.gapAttribute = gapAttribute;
  }

然后，我们可以将消息窗口化到新的自定义会话中：

.apply("Window into sessions", Window.<String>into(DynamicSessions
  .withDefaultGapDuration(Duration.standardSeconds(10))
  .withGapAttribute("gap"))

为了对此进行测试，我们将使用一个带有受控输入的简单示例。对于我们的用例，我们将根据运行应用程序的设备考虑用户的不同需求。桌面用户可以长时间闲置（允许更长的会话），而我们只希望移动用户的会话时间较短。我们生成一些模拟数据，其中一些消息包含gap 属性，而另一些则省略它（这些消息将使用默认窗口）：

.apply("Create data", Create.timestamped(
            TimestampedValue.of("{\"user\":\"mobile\",\"score\":\"12\",\"gap\":\"5\"}", new Instant()),
            TimestampedValue.of("{\"user\":\"desktop\",\"score\":\"4\"}", new Instant()),
            TimestampedValue.of("{\"user\":\"mobile\",\"score\":\"-3\",\"gap\":\"5\"}", new Instant().plus(2000)),
            TimestampedValue.of("{\"user\":\"mobile\",\"score\":\"2\",\"gap\":\"5\"}", new Instant().plus(9000)),
            TimestampedValue.of("{\"user\":\"mobile\",\"score\":\"7\",\"gap\":\"5\"}", new Instant().plus(12000)),
            TimestampedValue.of("{\"user\":\"desktop\",\"score\":\"10\"}", new Instant().plus(12000)))
        .withCoder(StringUtf8Coder.of()))

视觉上：

对于桌面用户，只有两个间隔 12 秒的事件。没有指定间隔，因此默认为 10 秒，并且两个分数不会相加，因为它们属于不同的会话。

另一个用户，mobile，有 4 个事件，分别间隔 2、7 和 3 秒。没有一个时间间隔大于默认间隔，因此对于标准会话，所有事件都属于一个单独的会话，加分为 18：

user=desktop, score=4, window=[2019-05-26T13:28:49.122Z..2019-05-26T13:28:59.122Z)
user=mobile, score=18, window=[2019-05-26T13:28:48.582Z..2019-05-26T13:29:12.774Z)
user=desktop, score=10, window=[2019-05-26T13:29:03.367Z..2019-05-26T13:29:13.367Z)

对于新会话，我们为这些事件指定 5 秒的“间隙”属性。第三条消息在第二条消息之后 7 秒出现，因此现在进入不同的会话。上一场 18 分的大型比赛将分为两个 9 分的比赛：

user=desktop, score=4, window=[2019-05-26T14:30:22.969Z..2019-05-26T14:30:32.969Z)
user=mobile, score=9, window=[2019-05-26T14:30:22.429Z..2019-05-26T14:30:30.553Z)
user=mobile, score=9, window=[2019-05-26T14:30:33.276Z..2019-05-26T14:30:41.849Z)
user=desktop, score=10, window=[2019-05-26T14:30:37.357Z..2019-05-26T14:30:47.357Z)

完整代码here。使用 Java SDK 2.13.0 测试

Python 版本：

类似地，我们可以将相同的方法扩展到 Python SDK。 Sessions 类的代码可以在 here 找到。我们将定义一个新的DynamicSessions 类。在assign方法中，我们可以使用context.element访问处理过的记录，并根据数据修改gap。

原文：

def assign(self, context):
  timestamp = context.timestamp
  return [IntervalWindow(timestamp, timestamp + self.gap_size)]

扩展：

def assign(self, context):
  timestamp = context.timestamp

  try:
    gap = Duration.of(context.element[1][“gap”])
  except:
    gap = self.gap_size

  return [IntervalWindow(timestamp, timestamp + gap)]

如果输入数据包含gap 字段，它将使用它来覆盖默认间隙大小。在我们的管道代码中，我们只需要将事件窗口化到DynamicSessions，而不是标准的Sessions：

'user_session_window'   >> beam.WindowInto(DynamicSessions(gap_size=gap_size),
                                             timestamp_combiner=window.TimestampCombiner.OUTPUT_AT_EOW)

使用标准会话，输出如下：

INFO:root:>> User mobile had 4 events with total score 18 in a 0:00:22 session
INFO:root:>> User desktop had 1 events with total score 4 in a 0:00:10 session
INFO:root:>> User desktop had 1 events with total score 10 in a 0:00:10 session

使用我们的自定义窗口移动事件分为两个不同的会话：

INFO:root:>> User mobile had 2 events with total score 9 in a 0:00:08 session
INFO:root:>> User mobile had 2 events with total score 9 in a 0:00:07 session
INFO:root:>> User desktop had 1 events with total score 4 in a 0:00:10 session
INFO:root:>> User desktop had 1 events with total score 10 in a 0:00:10 session

所有文件here。使用 Python SDK 2.13.0 测试

【讨论】：