如何使用 Apache Beam 模型导航树
How to navigate a tree using Apache Beam model
我有一个开始接收类别 ID 列表的管道。
在 ParDo 中，我执行一个 DoFn，它使用这些 ID 作为参数调用 REST API，并且 return 是一个 PCollection 15=] object.
.apply("Read Category", ParDo.of(new DoFn<String, Category>(){});
一秒钟后 ParDo 我坚持这个 Category objects，读取他的 children 属性和 return 他的 children ID。
.apply("Persist Category", ParDo.of(new DoFn<Category, String>(){});
我想在由第二个 ParDo 编辑的 ID 列表 return 中再次重复第一个 ParDo，直到没有 children 个类别。
如何使用受益于并行处理的 Apache Beam 模型执行此操作？
Apache Beam 目前不提供任何用于迭代并行处理的原语。您可以使用一些解决方法，例如其中一些列在 .
中
另一种方法是编写一个简单的 Java 函数，该函数将遍历特定顶级 ID 的树（从给定 ID 开始递归获取类别和子级），并使用 ParDo 应用该函数并行 - 但是，显然，该函数内不会有分布式并行性。
您也可以先部分 "unroll" 管道中的迭代，以获得跨树的前几层的一组分布式并行性 - 例如用第一和第二个 ParDo 的序列构建一个管道，然后应用第三个 ParDo，它应用迭代 Java 函数来遍历剩余的级别。
请注意，如果您在 Dataflow 或任何其他支持融合优化的运行器上执行，很可能您需要使用 preventing fusion 的技巧之一。
我有一个开始接收类别 ID 列表的管道。
在 ParDo 中，我执行一个 DoFn，它使用这些 ID 作为参数调用 REST API，并且 return 是一个 PCollection 15=] object.
.apply("Read Category", ParDo.of(new DoFn<String, Category>(){});
一秒钟后 ParDo 我坚持这个 Category objects，读取他的 children 属性和 return 他的 children ID。
.apply("Persist Category", ParDo.of(new DoFn<Category, String>(){});
我想在由第二个 ParDo 编辑的 ID 列表 return 中再次重复第一个 ParDo，直到没有 children 个类别。
如何使用受益于并行处理的 Apache Beam 模型执行此操作？
Apache Beam 目前不提供任何用于迭代并行处理的原语。您可以使用一些解决方法，例如其中一些列在
另一种方法是编写一个简单的 Java 函数，该函数将遍历特定顶级 ID 的树（从给定 ID 开始递归获取类别和子级），并使用 ParDo 应用该函数并行 - 但是，显然，该函数内不会有分布式并行性。
您也可以先部分 "unroll" 管道中的迭代，以获得跨树的前几层的一组分布式并行性 - 例如用第一和第二个 ParDo 的序列构建一个管道，然后应用第三个 ParDo，它应用迭代 Java 函数来遍历剩余的级别。
请注意，如果您在 Dataflow 或任何其他支持融合优化的运行器上执行，很可能您需要使用 preventing fusion 的技巧之一。