python - 确保用户 session 数据在 apache beam 中的多个 worker 上正确聚合

Question

假设我将以下 session 数据分配给两个工作人员:

+---------+-----------+--------+
| user_id | timestamp | worker |
+---------+-----------+--------+
|       1 |         2 |      1 |
|       1 |         1 |      1 |
|       1 |        21 |      1 |
|       1 |        23 |      2 |
|       1 |        24 |      2 |
+---------+-----------+--------+

我想确保如果我聚合数据，我会得到以下结果:

+---------+---------------+-------------+---------------+
| user_id | session_start | session_end | element_count |
+---------+---------------+-------------+---------------+
|       1 |             1 |           2 |             2 |
|       1 |            21 |          24 |             3 |
+---------+---------------+-------------+---------------+

而不是以下结果:

+---------+---------------+-------------+---------------+
| user_id | session_start | session_end | element_count |
+---------+---------------+-------------+---------------+
|       1 |             1 |           2 |             2 |
|       1 |            21 |          21 |             1 |
|       1 |            23 |          24 |             2 |
+---------+---------------+-------------+---------------+

例如 this用例，在将输入拆分到 session 窗口之前，我必须按用户 ID 对输入进行分区，以获取用户的所有 session 数据。

我的问题是:

• Apache Beam 是否自动确保每个工作人员都拥有所有数据？
• 例如，此代码片段是否确保所有用户 session 都正确聚合，即使数据被拆分给多个工作人员？

import apache_beam as beam
from apache_beam.transforms.window import Session

(pcoll 
 | beam.WindowInto(Session(gap_size=10))
 | beam.GroupByKey())

• 如果是这样，它究竟如何保证这种行为？
• 如何在本地复制这种特定情况以便进行测试？
• 我是否可以在文档中阅读有关此行为的更多信息？

Answer 1

Does Apache Beam automatically make sure that each worker has all the data? For example, does this code snippet make sure that all user sessions are agreggated correctly, even if the data was split over multiple workers?

Apache Beam 语义不依赖于哪个数据由哪个 worker 处理。由特定的 Beam runner 来保证它正确地执行流水线。如果某个运行者需要收集一个 worker 的所有数据来做这件事——那么那个运行者应该这样做；如果运行者无需这样做就可以保证正确性 - 甚至更好。

If so, how exactly does it guarantee this behaviour?

有效评估窗口和触发器的具体实现细节非常复杂。你可以看看ReduceFnRunner中的java实现.但是在较高的层次上，你可以这样想(这只是一个心智模型，不是精确的实现):

• 窗口转换根据其时间戳将每个元素分配到一个或多个窗口。
• GroupByKey 转换按键和窗口隐式分组，即对于每个键/窗口对，它维护一个包含此键和窗口的元素缓冲区。
• 如果 windowfn 正在“合并”(即，如果具有相同键的不同元素的窗口不是独立的 - 就像 session 窗口的情况一样)，它会为所有具有该键的元素重新计算正确的窗口集，并且将每个元素重新映射到各自的新窗口中。 (mergeWindows)

How can I replicate this specific situation locally so I can test it?

使用在本地内存中运行管道的直接运行器，并提供 Beam 模型的引用实现。