我想知道为什么有#sideInput()方法移至ProcessContext班级？
以前我可以在#startBundle()方法并缓存结果。
这样做在#processElement()听起来效率较低。当然，我可以在将数据传递到视图之前进行预处理，但是仍然存在调用的开销#sideInput()对于每个元素...
谢谢，
G
很好的问题。原因是我们添加了对窗口 PCollections 作为侧面输入的支持。这支持其他场景，包括在流模式下使用带有无界 PCollection 的侧输入。
在更改之前，我们仅支持全局窗口化的侧输入，然后在处理主输入 PCollection 的每个元素时，整个侧输入 PCollection 都可用。这对于传统批处理样式处理中的有界 PCollection 效果很好，但不能扩展到窗口或无界 PCollection。
更改后，您在 ParDo 中处理的当前元素的窗口控制侧输入的哪些子集可见。 （因此您无法访问 startBundle() 中的侧面输入，其中没有当前元素，因此也没有当前窗口。）
例如，考虑一个示例，其中您有一个流处理管道处理您的网站日志并向实时使用仪表板提供实时更新。您有两个无限制的输入 PCollection：一个包含新用户注册，另一个包含用户点击。您可以通过按小时对两个 PCollection 进行窗口化并对用户点击执行 ParDo（将新用户注册作为辅助输入）来识别哪些用户点击来自新用户。现在，当您处理给定小时内的用户点击时，您会自动看到同一小时内新用户注册的子集。您可以通过更改窗口函数并在侧面输入上及时向前移动元素时间戳来对此进行不同的变体 - 例如继续窗口用户每小时的点击次数，但使用过去 24 小时的新注册。
我确实同意这一更改使得在您的侧面输入上缓存任何后处理变得更加困难。我们添加了 View.asMultimap 来处理将 Iterable 转换为查找表的常见情况。如果您的后处理是按元素进行的，则可以在创建 PCollectionView 之前使用 ParDo 来完成。对于现在的其他事情，我建议从 processElement 中延迟执行。我有兴趣了解其他发生的模式，这样我们就可以设法提高它们的效率。
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)
