在 Beam Python DoFn 中进行昂贵的一次性初始化的推荐方法是什么？ Java SDK 有 DoFn.Setup，但在 Beam Python 中似乎没有等效项。
目前在 DoFn 初始化器中 attach objects to threading.local() 的最佳方式是什么？
【问题讨论】：
标签： google-cloud-dataflow apache-beam
数据流 Python 对于初始化昂贵对象的最佳方法并不是特别透明。有一些机制可以不频繁地实例化对象（目前只执行一次初始化并不理想）。下面概述了我进行的一些实验和得出的结论。希望 Beam 社区的人可以帮助纠正我的错误。
__init__
虽然__init__ 方法可以用来初始化一个昂贵的对象，但这种初始化不会发生在 Worker 机器上。该对象需要被序列化才能发送给 Worker，对于大型对象以及 Tensorflow 模型，它可能非常笨重或根本无法工作。此外，由于该对象将被序列化并通过线路发送，因此在此处执行初始化并不安全，因为可以拦截有效负载。建议反对使用这种方法。
start_bundle()
Dataflow 处理离散组中的数据，称为捆绑包。这些在批处理中定义得相当好，但在流式处理中它们取决于吞吐量。没有配置 Dataflow 如何创建其捆绑包的机制，实际上捆绑包的大小完全由 Dataflow 决定。 start_bundle() 方法将在 Worker 上调用，可用于初始化状态，但是实验发现，在流式上下文中，此方法的调用频率比预期的要高，并且会经常发生昂贵的重新初始化。
延迟初始化
这种方法是由 Beam 文档建议的，令人惊讶的是它的性能最高。延迟初始化是指创建一些有状态的参数，初始化为None，然后执行如下代码：
if self.expensive_object is None:
self.expensive_object = self.__expensive_initialization()
您可以直接在您的process() 方法中执行此代码。您还可以轻松地将一些依赖于global 状态的辅助函数组合在一起，这样您就可以拥有诸如（本文底部的一个示例）：
self.expensive_object = get_or_initialize_global(‘expensive_object’, self.__expensive_initialization)
实验
以下实验是在使用start_bundle 和上述惰性初始化方法配置的作业上运行的，并使用适当的日志记录来指示调用。各种吞吐量被发布到适当的队列，并相应地记录结果。
在 100 秒内以 1 条消息/秒的速率：
Context Number of Invocations
------------------------------------------------------------
NEW BUNDLE 100
LAZY INITIALIZATION 25
TOTAL MESSAGES 100
在 100 秒内以 10 msg/sec 的速率
Context Number of Invocations
------------------------------------------------------------
NEW BUNDLE 942
LAZY INITIALIZATION 3
TOTAL MESSAGES 1000
在 100 秒内以 100 条消息/秒的速率
Context Number of Invocations
------------------------------------------------------------
NEW BUNDLE 2447
LAZY INITIALIZATION 30
TOTAL MESSAGES 10000
在 100 秒内以 1000 msg/秒的速率
Context Number of Invocations
------------------------------------------------------------
NEW BUNDLE 2293
LAZY INITIALIZATION 36
TOTAL MESSAGES 100000
要点
尽管start_bundle 非常适合高吞吐量，但不管吞吐量如何，延迟初始化仍然是性能最高的。这是在 Python Beam 上执行昂贵初始化的推荐方法。鉴于官方文档中的以下引用，这个结果可能并不令人惊讶：
Setup - 每个 DoFn 实例先调用一次；这尚未在 Python SDK 中实现，因此用户只需使用延迟初始化即可解决问题
不过，被称为“变通”的事实并不是特别令人鼓舞，也许我们可以期待在不久的将来会有更强大的东西。
代码示例
由 Andreas Jansson 提供：
def get_or_initialize_global(object_key, initialize_expensive_object):
if object_key in globals():
expensive_object = globals()[object_key]
else:
expensive_object = initialize_expensive_object()
globals()[object_key] = expensive_object
【讨论】：
@property 或@lru_cache 装饰的方法内部的惰性初始化？
设置和拆卸现在是added to the Python SDK，并且是在 Beam Python DoFn 中进行昂贵的一次性初始化的推荐方法。
【讨论】：
【讨论】：
start_bundle 函数在一组项目的开头运行，但在 DoFn 生命周期内执行多次。在 Java SDK 中有 Setup()、StartBundle() 和 Process()（这里讨论区别：stackoverflow.com/a/50068377/135797），但 Python 只有 start_bundle() 和 process()。
apache_beam==2.4.0 源代码中搜索“拆卸”和“设置”也没有任何希望，所以它可能还不支持。在内部数据流运行器库中提到了 Teardown Policy：github.com/apache/beam/blob/master/sdks/python/apache_beam/… 希望它很快就会对我们有用。