google-cloud-dataflow - 如何解决 apache_beam.internal.clients.dataflow.dataflow_v1b3_messages.TypeValueValuesEnum 类上的酸洗错误？

Question

当我远程运行数据管道时，会引发 PicklingError:数据管道是使用适用于 Python 的 Beam SDK 编写的，并且我在 Google Cloud Dataflow 之上运行它。当我在本地运行时，管道工作正常。

以下代码生成 PicklingError:这应该会重现问题

import apache_beam as beam
from apache_beam.transforms import pvalue
from apache_beam.io.fileio import _CompressionType
from apache_beam.utils.options import PipelineOptions
from apache_beam.utils.options import GoogleCloudOptions
from apache_beam.utils.options import SetupOptions
from apache_beam.utils.options import StandardOptions

if __name__ == "__main__":
  pipeline_options = PipelineOptions()
  pipeline_options.view_as(StandardOptions).runner = 'BlockingDataflowPipelineRunner'
  pipeline_options.view_as(SetupOptions).save_main_session = True
  google_cloud_options = pipeline_options.view_as(GoogleCloudOptions)
  google_cloud_options.project = "project-name"
  google_cloud_options.job_name = "job-name"
  google_cloud_options.staging_location = 'gs://path/to/bucket/staging'
  google_cloud_options.temp_location = 'gs://path/to/bucket/temp'
  p = beam.Pipeline(options=pipeline_options)
  p.run()

下面是回溯开始和结束的示例:

WARNING: Could not acquire lock C:\Users\ghousains\AppData\Roaming\gcloud\credentials.lock in 0 seconds
WARNING: The credentials file (C:\Users\ghousains\AppData\Roaming\gcloud\credentials) is not writable. Opening in read-only mode. Any refreshed credentials will only be valid for this run.
Traceback (most recent call last):
  File "formatter_debug.py", line 133, in <module>
    p.run()
  File "C:\Miniconda3\envs\beam\lib\site-packages\apache_beam\pipeline.py", line 159, in run
    return self.runner.run(self)
    ....
    ....
    ....
  File "C:\Miniconda3\envs\beam\lib\sitepackages\apache_beam\runners\dataflow_runner.py", line 172, in run
    self.dataflow_client.create_job(self.job))    
  StockPickler.save_global(pickler, obj)
  File "C:\Miniconda3\envs\beam\lib\pickle.py", line 754, in save_global (obj, module, name)) 
  pickle.PicklingError: Can't pickle <class 'apache_beam.internal.clients.dataflow.dataflow_v1b3_messages.TypeValueValuesEnum'>: it's not found as apache_beam.internal.clients.dataflow.dataflow_v1b3_messages.TypeValueValuesEnum

Answer 1

我发现当 Pipeline 对象包含在被腌制并发送到云的上下文中时，会引发错误:

pickle.PicklingError: Can't pickle <class 'apache_beam.internal.clients.dataflow.dataflow_v1b3_messages.TypeValueValuesEnum'>: it's not found as apache_beam.internal.clients.dataflow.dataflow_v1b3_messages.TypeValueValuesEnum

当然，你可能会问:

1. 为什么 Pipeline 对象在发送到云端时变得不可pickle，因为通常它是可 pickle 的？
2. 如果这确实是问题所在，那么我不会一直收到此错误 - 发送到云的上下文中通常不包含 Pipeline 对象吗？
3. 如果 Pipeline 对象通常不包含在发送到云端的上下文中，那么为什么我的案例中包含 Pipeline 对象？

(1)

当您调用p.run()时在管道上 cloud=True ，首先发生的事情之一是 p.runner.job=apiclient.Job(pipeline.options)设置于apache_beam.runners.dataflow_runner.DataflowPipelineRunner.run .

如果没有设置此属性，管道是可pickle的。但是一旦设置了这个，管道就不再是可腌制的，因为 p.runner.job.proto._Message__tags[17]是 TypeValueValuesEnum ，它在 apache_beam.internal.clients.dataflow.dataflow_v1b3_messages 中定义为嵌套类。据我所知，嵌套类不能被腌制(即使是通过 dill - 请参阅 How can I pickle a nested class in python? )。

(2)-(3)

与直觉相反，Pipeline 对象通常不包含在发送到云的上下文中。当您调用p.run()时在管道上 cloud=True ，只有以下对象被腌制(请注意，腌制发生在 p.runner.job 设置之后):

1. 如果save_main_session=True ，则模块中的所有全局对象指定 __main__被腌制的。 ( __main__ 是您从命令行运行的脚本)。
2. 管道中定义的每个转换都是单独腌制的

就您的情况而言，您遇到了#1，这就是您的解决方案有效的原因。我实际上遇到了#2，我定义了 beam.Map lambda 函数作为复合方法 PTransform 。 (当应用复合转换时，管道将作为转换的属性添加...)我的解决方案是在模块中定义这些 lambda 函数。

我们的长期解决方案是在 Apache Beam 项目中修复此问题。待定!