数据流：在事件流中查找上一个事件

Question

在 Google Dataflow 中恢复我正在寻找的与 Apache Beam 相关的操作类似于 Azure 流分析中的 LAG

使用我接收数据的 X 分钟窗口：

||||||  ||||||  ||||||  ||||||  ||||||  ||||||
|  1 |  |  2 |  |  3 |  |  4 |  |  5 |  |  6 | 
|id=x|  |id=x|  |id=x|  |id=x|  |id=x|  |id=x| 
|||||| ,|||||| ,|||||| ,|||||| ,|||||| ,|||||| , ...

我需要将 data(n) 与 data(n-1) 进行比较，例如，按照前面的示例，它将是这样的：

if data(6) inside and data(5)  outside then ... 
if data(5) inside and data(4)  outside then ... 
if data(4) inside and data(3)  outside then ... 
if data(3) inside and data(2)  outside then ... 
if data(2) inside and data(1)  outside then ... 

有什么“实用”的方法可以做到这一点吗？

Answer 1

使用 Beam，如 docs 中所述，每个键和窗口都保持状态。因此，您无法访问以前窗口中的值。

要完成您想做的事情，您可能需要更复杂的管道设计。我的想法（在这里作为示例开发）是在 ParDo 中复制您的消息：

• 将它们不加修改地发送到主输出
• 同时，将它们发送到具有单窗口延迟的侧面输出

要执行第二个要点，我们可以将窗口的持续时间 (WINDOW_SECONDS) 添加到元素时间戳：

class DuplicateWithLagDoFn(beam.DoFn):

  def process(self, element, timestamp=beam.DoFn.TimestampParam):
    # Main output gets unmodified element
    yield element
    # The same element is emitted to the side output with a 1-window lag added to timestamp
    yield beam.pvalue.TaggedOutput('lag_output', beam.window.TimestampedValue(element, timestamp + WINDOW_SECONDS))

我们调用指定正确标签的函数：

beam.ParDo(DuplicateWithLagDoFn()).with_outputs('lag_output', main='main_output')

然后将相同的窗口方案应用于两者，按键共同分组等。

windowed_main = results.main_output | 'Window main output' >> beam.WindowInto(window.FixedWindows(WINDOW_SECONDS))
windowed_lag = results.lag_output | 'Window lag output' >> beam.WindowInto(window.FixedWindows(WINDOW_SECONDS))

merged = (windowed_main, windowed_lag) | 'Join Pcollections' >> beam.CoGroupByKey()

最后，我们可以在同一个 ParDo 中同时拥有两个值（旧的和新的）：

class CompareDoFn(beam.DoFn):

  def process(self, element):
    logging.info("Combined with previous vale: {}".format(element))

    try:
      old_value = int(element[1][1][0].split(',')[1])
    except:
      old_value = 0

    try:
      new_value = int(element[1][0][0].split(',')[1])
    except:
      new_value = 0

    logging.info("New value: {}, Old value: {}, Difference: {}".format(new_value, old_value, new_value - old_value))
    return (element[0], new_value - old_value)

为了测试这一点，我使用直接运行器运行管道，并在单独的 shell 上发布两条消息，间隔超过 10 秒（在我的情况下，WINDOW_SECONDS 是 10 秒）：

gcloud pubsub topics publish lag --message="test,120"
sleep 12
gcloud pubsub topics publish lag --message="test,40"

并且作业输出显示了预期的差异：

INFO:root:New message: (u'test', u'test,120')
INFO:root:Combined with previous vale: (u'test', ([u'test,120'], []))
INFO:root:New value: 120, Old value: 0, Difference: 120
INFO:root:New message: (u'test', u'test,40')
INFO:root:Combined with previous vale: (u'test', ([u'test,40'], [u'test,120']))
INFO:root:New value: 40, Old value: 120, Difference: -80
INFO:root:Combined with previous vale: (u'test', ([], [u'test,40']))
INFO:root:New value: 0, Old value: 40, Difference: -40

我的示例 here 的完整代码。在复制元素时要考虑性能因素，但如果您需要在两个窗口期间提供可用的值，这很有意义。