DataflowRunner运行报错‘Clients have non-trivial state...’求助

首先得明确：DirectRunner和DataflowRunner的运行机制完全不同，哪怕你指定了相同的依赖版本，实际执行时的序列化要求也天差地别。

核心差异原因 #

• DirectRunner是在你本地的单个进程里跑完全部流水线，对象大多是在同一个进程内复用，很多时候不需要严格的序列化（pickle），就算有些对象不可pickle，也能正常工作；
• DataflowRunner则是把你的代码打包，分发到云端的多个Worker节点上执行。所有需要跨进程、跨节点传递的对象（包括你的DoFn实例）都必须能被pickle序列化。哪怕你在start_bundle里初始化客户端，DoFn本身还是会被序列化后传到Worker，一旦DoFn里有任何不可pickle的属性（哪怕是后续才赋值的），就会触发你遇到的报错。

你尝试的方案为什么没解决问题？ #

你用start_bundle初始化客户端的思路是对的，但在老旧的Beam 2.3.0版本里，这个方法的执行时机和序列化逻辑可能有问题——DoFn实例在被传到Worker之前还是会被尝试序列化，而某些旧版本的google-cloud客户端（比如你用的1.6.0版本的datastore客户端）本身就带有无法被pickle的内部状态，哪怕你是在start_bundle里才赋值，也可能被序列化逻辑误触发。

可行的解决方案 #

1. 用线程本地存储避免客户端被序列化

修改你的DoFn，把客户端放在线程本地存储里，确保它只会在Worker进程的线程内初始化，绝不会被序列化传递：

import threading
from google.cloud import datastore

class MyDoFn(beam.DoFn):
    def __init__(self):
        self._local = threading.local()

    @property
    def dsclient(self):
        if not hasattr(self._local, 'client'):
            self._local.client = datastore.Client()
        return self._local.client

    def process(self, context):
        # 用self.dsclient来操作数据存储
        key = self.dsclient.key('EntityKind', context.element['id'])
        entity = self.dsclient.get(key)
        # 后续业务逻辑

2. 升级Apache Beam和依赖版本（强烈推荐）

你用的Beam 2.3.0是2018年的旧版本，后续的Beam版本（比如2.40+及以后）针对云客户端的序列化问题做了大量修复，比如引入了更可靠的setup方法做初始化，还优化了DoFn的序列化逻辑。同时同步升级google-cloud相关依赖到兼容的新版本，能从根本上避免这类pickle问题。

3. 确保Dataflow使用正确的依赖

提交Dataflow作业时，一定要通过--requirements_file=requirements.txt参数明确指定依赖文件，或者用setup.py来管理依赖，防止云端Worker使用的依赖版本和本地不一致——有时候哪怕你本地指定了版本，Dataflow默认可能会拉取其他版本的依赖。

内容的提问来源于stack exchange，提问作者user9773014