当我远程运行数据管道时，会引发 PicklingError：数据管道是使用 Beam SDK for Python 编写的，并且我在 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
我发现当 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
自然，你可能会问：
- 既然 Pipeline 对象通常是可pickle的，那么当它被发送到云时，是什么使得它变得不可pickle呢？
- 如果这确实是问题所在，那么我不会一直收到此错误 - Pipeline 对象通常不是包含在发送到云的上下文中吗？
- 如果 Pipeline 对象通常不包含在发送到云的上下文中，那么为什么 Pipeline 对象包含在我的案例中？
(1)
你打电话时p.run()在管道上cloud=True，首先发生的事情之一是p.runner.job=apiclient.Job(pipeline.options)安顿好了apache_beam.runners.dataflow_runner.DataflowPipelineRunner.run.
如果没有设置此属性，则 Pipeline 是可 pickle 的。但是一旦设置完毕，管道就不再是可腌制的，因为p.runner.job.proto._Message__tags[17] is a TypeValueValuesEnum，它被定义为嵌套类apache_beam.internal.clients.dataflow.dataflow_v1b3_messages。据我所知，嵌套类不能被腌制（即使是通过莳萝 - 请参阅如何在 python 中 pickle 嵌套类？ https://stackoverflow.com/questions/1947904/how-can-i-pickle-a-nested-class-in-python).
(2)-(3)
与直觉相反，Pipeline 对象通常不包含在发送到云的上下文中。你打电话时p.run()在管道上cloud=True，只有以下对象被腌制（请注意，腌制发生在p.runner.job已设置）：
- If
save_main_session=True，则指定模块中的所有全局对象__main__被腌制的。 （__main__是您从命令行运行的脚本）。
- 管道中定义的每个转换都是单独腌制的
就您而言，您遇到了#1，这就是您的解决方案有效的原因。我实际上遇到了#2，我定义了一个beam.Maplambda 函数作为复合方法PTransform。 （当应用复合转换时，管道将作为转换的属性添加...）我的解决方案是在模块中定义这些 lambda 函数。
一个长期的解决方案是我们在 Apache Beam 项目中解决这个问题。待定！
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)
