如何解决类 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()

以下是 Traceback 开头和结尾的示例：

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 对象在发送到云端时不可腌制，因为通常它是可腌制的？
2. 如果这确实是问题所在，那么我不会一直收到此错误 - 通常情况下，Pipeline 对象不是包含在发送到云的上下文中吗？
3. 如果 Pipeline 对象通常不包含在发送到云端的上下文中，那么为什么我的案例中包含 Pipeline 对象？

(1)

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

没有这个属性集，管道是可腌制的。但是一旦设置了这个，管道就不再是可腌制的，因为p.runner.job.proto._Message__tags[17] 是一个TypeValueValuesEnum，它在apache_beam.internal.clients.dataflow.dataflow_v1b3_messages 中被定义为一个嵌套类。无法腌制 AFAIK 嵌套类（即使是 dill - 请参阅 How can I pickle a nested class in python?）。

(2)-(3)

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

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

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

一个长期的解决方案是让我们在 Apache Beam 项目中解决这个问题。待定！

Answer 2

这应该在 google-dataflow 0.4.4 sdk 版本中使用https://github.com/apache/incubator-beam/pull/1485 修复

Answer 3

我通过将 main 的主体封装在 run() 方法中并调用 run() 解决了这个问题。