跨多个工作人员的输入一致性答案

【问题标题】：sideInput consistensy across multiple workers跨多个工作人员的输入一致性
【发布时间】：2017-01-07 22:40:51
【问题描述】：

我正在研究需要将控制数据广播到 DoFn 转换的每个实例的管道。理想情况下，我想获得所有这些控制数据，而不仅仅是最后一个状态。我将示例简化为非常简单的示例——为侧输入和主输入设置两个 CountingInput，为侧输入过滤 30 个第一个刻度并寻找侧输入。

    PCollection<Long> iDs =
            p.apply(CountingInput.unbounded().withRate(1, Duration.millis(200)))
                    .apply(ParDo.of(new DoFn<Long, Long>() {
                        @Override
                        public void processElement(ProcessContext c) {
                           Long cnt = c.element();
                           if (cnt <= 30) {
                               logger.info("ID=" + cnt);
                               c.output(cnt);
                           }
                        }
                    }));

    PCollectionView<List<Long>> iDsView = iDs
            .apply(Window.<Long>into(new GlobalWindows())
                    .triggering(Repeatedly.forever(AfterPane.elementCountAtLeast(1)))
                    .discardingFiredPanes()
            )
            .apply(View.asList());

    p.apply(CountingInput.unbounded().withRate(1, Duration.millis(1000)))
            .apply(ParDo
                    .withSideInputs(iDsView)
                    .of(new DoFn<Long, String>() {
                        @Override
                        public void processElement(ProcessContext c) {
                            Long in = c.element();
                            List<Long> si = c.sideInput(iDsView);

                            StringBuilder sb = new StringBuilder();
                            si.forEach(x -> sb.append(",").append(x));
                            logger.info("invocation=" + in
                                    + " class=" + this.toString()
                                    + " sideInput=[" + sb.toString().substring(1) + "]");
                        }
                    }));

使用 InProcessPipelineRunner 我最终会看到类似

的输出

INFO: invocation=10 class=com.sandbox.dw.WriteLogsToBQ$2@221a6d4a sideInput=[30]
Jan 08, 2017 12:38:44 AM com.sandbox.dw.WriteLogsToBQ$2 processElement
INFO: invocation=11 class=com.sandbox.dw.WriteLogsToBQ$2@449a2a7a sideInput=[30]
Jan 08, 2017 12:38:45 AM com.sandbox.dw.WriteLogsToBQ$2 processElement
INFO: invocation=12 class=com.sandbox.dw.WriteLogsToBQ$2@4a289e60 sideInput=[30]

但是当我使用 --runner=BlockingDataflowPipelineRunner --numWorkers=4 运行它时我看到 4 名工作人员之间的 sideInlut 非常不一致，这在我运行管道的几分钟内是相同的：

    00:47:16.586
    invocation=138 class=com.sandbox.dw.WriteLogsToBQ$2@312aa182 sideInput=[0]
    00:47:15.709
    invocation=137 class=com.sandbox.dw.WriteLogsToBQ$2@2d0b6481 sideInput=[3,6,9,12,18,21,24,30]
    00:47:14.445
    invocation=136 class=com.sandbox.dw.WriteLogsToBQ$2@5153b895 sideInput=[0]
    00:47:11.760
    invocation=134 class=com.sandbox.dw.WriteLogsToBQ$2@65683230 sideInput=[3,6,9,12,18,21,24,30]
    00:47:11.231
    invocation=132 class=com.sandbox.dw.WriteLogsToBQ$2@5ee8917a sideInput=[0]
    00:47:10.775
    invocation=133 class=com.sandbox.dw.WriteLogsToBQ$2@16000b0 sideInput=[3,6,9,12,18,21,24,30]
    00:47:09.477
    invocation=123 class=com.sandbox.dw.WriteLogsToBQ$2@6ffe3f47 sideInput=[15]
    00:47:08.977
    invocation=130 class=com.sandbox.dw.WriteLogsToBQ$2@458bc76b sideInput=[3,6,9,12,18,21,24,30]
    00:47:07.505
    invocation=129 class=com.sandbox.dw.WriteLogsToBQ$2@2c6fcbcf sideInput=[0]
    00:47:07.200
    invocation=128 class=com.sandbox.dw.WriteLogsToBQ$2@1bf63883 sideInput=[3,6,9,12,18,21,24,30]
    00:47:06.033
    invocation=127 class=com.sandbox.dw.WriteLogsToBQ$2@5fd02daf sideInput=[3,6,9,12,18,21,24,30]
    00:47:05.573
    invocation=119 class=com.sandbox.dw.WriteLogsToBQ$2@7ba4a88b sideInput=[15]
    00:47:04.502
    invocation=126 class=com.sandbox.dw.WriteLogsToBQ$2@a7d0a48 sideInput=[0]

我还注意到每个输入元素都会重新创建 DoFn 实例。任何人都可以建议使用 sideInput 保证将 PubSub 数据广播到每个转换的最佳方法吗？

这是一个简单的例子来分享我的担忧：

    PCollectionView<List<Long>> iDsView =
            p.apply(CountingInput.unbounded().withRate(1, Duration.millis(1000)))
                    .apply(Window.<Long>into(new GlobalWindows())
                            .triggering(Repeatedly.forever(AfterPane.elementCountAtLeast(1)))
                            .discardingFiredPanes()
                    )
                    .apply(Max.longsGlobally())
                    .apply(ParDo.of(new DoFn<Long, Long>() {
                        @Override
                        public void processElement(ProcessContext c) {
                            Long elem = c.element();
                            logger.info("MaxElement=" + elem);
                            c.output(elem);
                        }
                    }))
                    .apply(View.asList());

    p.apply(CountingInput.unbounded().withRate(1, Duration.millis(300)))
            .apply(ParDo
                    .withSideInputs(iDsView)
                    .of(new DoFn<Long, Long>() {
                        @Override
                        public void processElement(ProcessContext c) {
                            Long in = c.element();
                            List<Long> si = c.sideInput(iDsView);
                            StringBuilder sb = new StringBuilder();
                            si.forEach(x -> sb.append(",").append(x));
                            logger.info("MainInput=" + in
                                    + " sideInput=[" + sb.toString().substring(1) + "]");
                        }
                    }));

它在本地运行器上运行良好，但使用 BlockingDataflowPipelineRunner 不会使用单个工作器更新侧面输入。我可以看到触发器是如何触发的，并将函数 sendt 结果合并到日志中。但是侧面输入返回相同的值。

本地跑步者日志：

    INFO: MaxElement=6
    Jan 12, 2017 9:22:52 PM com.sandbox.dw.WriteLogsToBQ$2 processElement
    INFO: MainInput=21 sideInput=[6]
    Jan 12, 2017 9:22:52 PM com.sandbox.dw.WriteLogsToBQ$2 processElement
    INFO: MainInput=22 sideInput=[6]
    Jan 12, 2017 9:22:53 PM com.sandbox.dw.WriteLogsToBQ$2 processElement
    INFO: MainInput=23 sideInput=[6]
    Jan 12, 2017 9:22:53 PM com.sandbox.dw.WriteLogsToBQ$1 processElement
    INFO: MaxElement=7
    Jan 12, 2017 9:22:53 PM com.sandbox.dw.WriteLogsToBQ$2 processElement
    INFO: MainInput=24 sideInput=[7]
    Jan 12, 2017 9:22:53 PM com.sandbox.dw.WriteLogsToBQ$2 processElement
    INFO: MainInput=25 sideInput=[7]
    Jan 12, 2017 9:22:53 PM com.sandbox.dw.WriteLogsToBQ$2 processElement
    INFO: MainInput=26 sideInput=[7]
    Jan 12, 2017 9:22:54 PM com.sandbox.dw.WriteLogsToBQ$1 processElement
    INFO: MaxElement=8
    Jan 12, 2017 9:22:54 PM com.sandbox.dw.WriteLogsToBQ$2 processElement
    INFO: MainInput=27 sideInput=[8]
    Jan 12, 2017 9:22:54 PM com.sandbox.dw.WriteLogsToBQ$2 processElement
    INFO: MainInput=28 sideInput=[8]
    Jan 12, 2017 9:22:54 PM com.sandbox.dw.WriteLogsToBQ$2 processElement
    INFO: MainInput=29 sideInput=[8]
    Jan 12, 2017 9:22:55 PM com.sandbox.dw.WriteLogsToBQ$2 processElement
    INFO: MainInput=30 sideInput=[8]
    Jan 12, 2017 9:22:55 PM com.sandbox.dw.WriteLogsToBQ$1 processElement
    INFO: MaxElement=9
    Jan 12, 2017 9:22:55 PM com.sandbox.dw.WriteLogsToBQ$2 processElement
    INFO: MainInput=31 sideInput=[9]
    Jan 12, 2017 9:22:55 PM com.sandbox.dw.WriteLogsToBQ$2 processElement
    INFO: MainInput=32 sideInput=[9]
    Jan 12, 2017 9:22:56 PM com.sandbox.dw.WriteLogsToBQ$2 processElement
    INFO: MainInput=33 sideInput=[9]
    Jan 12, 2017 9:22:56 PM com.sandbox.dw.WriteLogsToBQ$1 processElement

数据流日志：

    MaxElement=61
    21:26:00.225
    MainInput=207 sideInput=[0]
    21:26:00.676
    MainInput=208 sideInput=[0]
    21:26:00.924
    MainInput=209 sideInput=[0]
    21:26:01.258
    MainInput=210 sideInput=[0]
    21:26:01.260
    MaxElement=62
    21:26:01.518
    MainInput=211 sideInput=[0]
    21:26:01.748
    MainInput=212 sideInput=[0]
    21:26:02.071
    MainInput=213 sideInput=[0]
    21:26:02.313
    MainInput=214 sideInput=[0]
    21:26:02.466
    MaxElement=63
    21:26:02.677
    MainInput=215 sideInput=[0]
    21:26:02.994
    MaxElement=64
    21:26:03.113
    MainInput=216 sideInput=[0]
    21:26:03.335
    MainInput=217 sideInput=[0]
    21:26:03.614
    MainInput=218 sideInput=[0]
    21:26:04.132
    MainInput=219 sideInput=[0]
    21:26:04.142
    MaxElement=65
    21:26:04.193
    MainInput=220 sideInput=[0]
    21:26:04.538
    MainInput=221 sideInput=[0]
    21:26:04.706
    MainInput=222 sideInput=[0]
    21:26:05.250
    MaxElement=66
    21:26:05.531
    MainInput=224 sideInput=[0]

我担心的是侧面输入显示陈旧的结果，我希望触发触发器时侧面输入缓存失效。

【问题讨论】：

标签： google-cloud-dataflow

【解决方案1】：

您将遇到主要输入元素之间的一致性问题和辅助输入上的多个触发。一致性模型非常松散：

每次触发都会导致所有工作人员最终使用输出的元素更新，没有截止日期或同步规则。
与此同时，在他们读取之前可能会发生另一个触发，因此可能永远不会看到特定的输出元素。

对于触发本身，请记住它也是不确定的：

值将在处理过程中的自然提交点输出，即使满足触发器的谓词的元素少于完整的捆绑包也是如此。这允许在存在合并窗口、重试、网络延迟和捆绑变化的情况下进行有效的可预测行为。
因此它被指定为在元素计数为至少 1 之后。即使触发器的谓词仅满足一个，也将一起触发一束已处理元素的整体。因此可以看到 0 到 30 的任何子集。

因此，您的管道布局最自然地适用于侧输入表示向某个最终值（可能是无限的，因此收敛可能永远持续）的不确定收敛的情况。相反，我认为您会希望通过CoGroupByKey 或Sam 的答案中建议的直接侧频道订阅正确加入。

根据您的用例的实际细节，可能还有其他使用自定义WindowFn 的更深奥的解决方案。主输入将阻塞，直到侧输入在匹配窗口中至少有一个触发元素。

最后，关于累积模式的快速说明：

根据您“希望获取所有控制数据，而不仅仅是最后一个状态”的声明，听起来您希望将discardingFiredPanes() 替换为accumulatingFiredPanes()，但这会导致侧面输入无限成长。
如果您使用窗口化来限制侧面输入的生命周期，那么accumulatingFiredPanes() 会更好地工作 - 所有工作人员最终都会收到所有控制消息，并且窗口的累积消息将在窗口到期时释放。但是不能保证主输入元素实际上到达的时间足够晚才能看到最终值！所以它仍然可能不是正确的方法。

【讨论】：

感谢您的回复！关于我的案例的更多细节：控制实际上是 BQ 表中的一些基本状态，并从 PubSub 慢慢改变下一个版本。如果工人的 jvm 可以相互通信，我会简单地广播这些消息。阅读每个工作人员的个人订阅对我来说似乎有点过头了，除了 PubSub 的 alpha 版本 Java 客户端之外，我找不到其他的。
如果控件只是基础并且不再更改，那么侧面输入上的一次触发触发器将起作用。如果基础也在发生变化，那么您将无法真正将新基础与应在其之上应用的更改同步。
它不应该严格同步，最终只要几分钟就完美了。窗口化可能会有所帮助——版本不相交，但问题是在新版本到来之前我们不知道 maxTimestamp。
感谢您提供的清晰示例，我正在研究它。
回圈：这是一个缓存错误，数据没有更新。现在应该已经解决了。

【解决方案2】：

根据您的一致性需求，您可以有一个广播控制消息的 Pub/Sub 主题的每个进程的静态订阅者。这将是一个在初始化时创建对该主题的随机订阅并开始收听该主题的对象。每当它收到一条消息时，它就会将消息的内容提供给 DoFn 的处理代码。

请注意，没有很好的方法可以确保清理这些订阅，因此如果您经常启动和停止管道，则需要一些定期清理的方法。

【讨论】：

感谢您的建议。但是您知道是否有适当的方法可以为此使用侧输入？我看到该文档几乎描述了我需要的内容，但没有给出任何示例：“如果侧面输入有多个触发器触发，Dataflow 将使用最新触发器触发的值。如果您将侧面输入与单个触发器一起使用，这将特别有用全局窗口并指定触发器。”
Kenn 整理了一个更全面的答案——如果您还有问题，请告诉我？

【解决方案3】：

在将构成 Dataflow 2.x 基础的 Beam 中，您可以使用状态更严格地控制缓慢变化的侧面输入尺寸。

对于空间，我将假设 Either 类型存在并且具有默认编码器。然后，如果您有一些类型 Key、MainInput 和 SideInput 以及您想要更新补充状态的任何自定义代码，那么实现您想要的有状态 DoFn 可能如下所示：

new DoFn<KV<Key, Either<MainInput, SideInputUpdate>>, KV<Key, Output>>() {

  @StateId("side")
  private final StateSpec<ValueState<SideInput>> detailsSpec =
      StateSpecs.value(SideInput.getCoder());

  @ProcessElement
  public void processElement(
      ProcessContext ctx,
      @StateId("side") ValueState<SideInput> sideState) {

    SideInput side = sideState.read();

    if (ctx.element().getValue().isRight()) {
      SideInputUpdate update = ctx.element().getValue().getRight();
      sideState.write(sideInput.applyUpdate(update));
    } else {
      MainInput element = ctx.element().getValue().getLeft();
      // do whatever you want to do with the element
    }
  }
}

要使用此DoFn，您需要将主输入和辅助输入注入PCollection<Either<MainInput, SideInputUpdate>>，如下所示：

PCollection<MainInput> mainInput = ...
PCollection<SideInputUpdate> sideInputUpdates = ...

PCollection<Either<MainInput, SideInputUpdate>> injectedMain = 
    mainInput.apply(MapElements.via(... in left ...));
PCollection<Either<MainInput, SideInputUpdate>> injectedSide = 
    sideInputUpdates.apply(MapElements.via(... in right ...);

PCollection<KV<Key, Either<MainInput, SideInputUpdate>>> =
    PCollectionList.of(injectedMain).and(injectedSide)
        .apply(Flatten.pCollections())
        .apply(WithKeys.of(...))

结果仍然不确定 - 您的侧输入更新和主输入元素没有排序保证。但延迟会更低且相当可预测，因为状态会在对该键和窗口进行任何进一步处理之前更新。

有关此新功能的更多信息，请参阅this blog post on the Beam blog。

【讨论】：

感谢肯恩！关于状态的另一个问题。在我的情况下，状态是不可序列化的，基本上我们有一个代码在运行时在 scala (currentMirror/ToolBox) 中编译成一个类。
你能把它作为一个新问题提出并链接到它吗？
当然，我实际上并没有完成上面的问题，也没有意识到我发布了它:-(