python - 使用 Python 的 GCP 数据流。 "AttributeError: Can' t 获取模块 '_JsonSink' 上的属性 'dataflow_worker.start'

Question

我是 GCP 数据流新手。

我尝试从GCP云存储中将文本文件(一行JSON字符串)读取为JSON格式，然后根据某些字段的值将其拆分并输出到GCP云存储(作为JSON字符串文本文件)。

这里是my code

但是，我在 GCP 数据流上遇到一些错误:

Traceback (most recent call last):
  File "main.py", line 169, in <module>
    run()
  File "main.py", line 163, in run
    shard_name_template='')
  File "C:\ProgramData\Miniconda3\lib\site-packages\apache_beam\pipeline.py", line 426, in __exit__
    self.run().wait_until_finish()
  File "C:\ProgramData\Miniconda3\lib\site-packages\apache_beam\runners\dataflow\dataflow_runner.py", line 1346, in wait_until_finish
    (self.state, getattr(self._runner, 'last_error_msg', None)), self)
apache_beam.runners.dataflow.dataflow_runner.DataflowRuntimeException: Dataflow pipeline failed. State: FAILED, Error:
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/dataflow_worker/batchworker.py", line 773, in run
    self._load_main_session(self.local_staging_directory)
  File "/usr/local/lib/python3.7/site-packages/dataflow_worker/batchworker.py", line 489, in _load_main_session
    pickler.load_session(session_file)
  File "/usr/local/lib/python3.7/site-packages/apache_beam/internal/pickler.py", line 287, in load_session
    return dill.load_session(file_path)
  File "/usr/local/lib/python3.7/site-packages/dill/_dill.py", line 410, in load_session
    module = unpickler.load()
  File "/usr/local/lib/python3.7/site-packages/dill/_dill.py", line 474, in find_class
    return StockUnpickler.find_class(self, module, name)
AttributeError: Can't get attribute '_JsonSink' on <module 'dataflow_worker.start' from '/usr/local/lib/python3.7/site-packages/dataflow_worker/start.py'>

我可以在本地运行此脚本，但当我尝试使用dataflowRunner时失败

请给我一些建议。

PS。 apache-beam 版本:2.15.0

[更新1]

我尝试@Yueyang Qiu建议，添加

pipeline_options.view_as(SetupOptions).save_main_session = True

提供的链接显示:

DoFn's in this workflow relies on global context (e.g., a module imported at module level)

这个link支持以上建议。

但是，出现了同样的错误。

所以，我在想我的_JsonSink(继承自filebasedsink.FileBasedSink)的实现是否错误或者需要添加其他内容。

任何意见都将不胜感激，谢谢大家!

Answer 1

您遇到了一个已知问题，目前(截至 2.17.0 版本)，Beam 不支持 Python 3 上主模块中的 super() 调用。请查看以下中的可能解决方案BEAM-6158 。 Udi 的答案是解决此问题的好方法，直到 BEAM-6158 得到解决，这样您就不必在 Python 2 上运行管道。