使用 Python 进行数据流流式处理，MatchContinuous 无法按预期工作答案

【问题标题】：Dataflow Streaming using Python, MatchContinuously not working as expected使用 Python 进行数据流流式处理，MatchContinuous 无法按预期工作
【发布时间】：2021-10-20 07:54:04
【问题描述】：

我尝试使用数据流 python 库，使用最近提供的此链接从存储桶中以流方式读取数据。

https://beam.apache.org/releases/pydoc/2.33.0/apache_beam.io.fileio.html?highlight=matchall#apache_beam.io.fileio.MatchContinuously

这是我正在使用的代码 sn-p，用于定期轮询存储桶。

(pipeline
         | 'Match Files' >> fileio.MatchContinuously(file_pattern="gs://xyz/abc/*.txt", interval=10.0, has_deduplication=True)
         | 'Read Matches' >> fileio.ReadMatches()
......

此代码在等待片刻后失败。有人可以帮助我了解我所缺少的吗？这是堆栈跟踪。

<PCollection[Read Matches/ParDo(_ReadMatchesFn).None] at 0x7fad0ccb7e80>
WARNING:root:Make sure that locally built Python SDK docker image has Python 3.8 interpreter.
Traceback (most recent call last):
  File "<stdin>", line 3, in <module>
  File "/usr/local/Caskroom/miniconda/base/lib/python3.8/site-packages/apache_beam/pipeline.py", line 586, in __exit__
    self.result = self.run()
  File "/usr/local/Caskroom/miniconda/base/lib/python3.8/site-packages/apache_beam/pipeline.py", line 565, in run
    return self.runner.run_pipeline(self, self._options)
  File "/usr/local/Caskroom/miniconda/base/lib/python3.8/site-packages/apache_beam/runners/direct/direct_runner.py", line 131, in run_pipeline
    return runner.run_pipeline(pipeline, options)
  File "/usr/local/Caskroom/miniconda/base/lib/python3.8/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 195, in run_pipeline
    self._latest_run_result = self.run_via_runner_api(
  File "/usr/local/Caskroom/miniconda/base/lib/python3.8/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 206, in run_via_runner_api
    return self.run_stages(stage_context, stages)
  File "/usr/local/Caskroom/miniconda/base/lib/python3.8/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 384, in run_stages
    stage_results = self._run_stage(
  File "/usr/local/Caskroom/miniconda/base/lib/python3.8/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 663, in _run_stage
    assert (runner_execution_context.watermark_manager.get_stage_node(
AssertionError: wrong timestamp for StageNode<inputs=['ref_PCollection_PCollection_3_split'],side_inputs=[].

【问题讨论】：

标签： python google-cloud-dataflow apache-beam

【解决方案1】：

通过检查日志，您的代码似乎在 Read Matches/ParDo(_ReadMatchesFn) 而不是 MatchContinuously 中失败。

这可能是由于您使用了DirectRunner，如果我没记错的话，StatefulDoFns 尚未完全支持(matrix)。 RemoveDuplicates 是一个有状态的DoFn。

我在 Dataflow 中测试了您的代码，它运行良好，没有任何问题（2.32 和 2.33）。

(p | MatchContinuously(f"gs://{bucket}/match/*", interval=10.0, has_deduplication=True)
   | ReadMatches()
)

另外，你并不需要ReadMatches，你可以这样做来阅读文本文件。也许你可以测试一下，看看它在 DirectRunner 中是否有效

(p | MatchContinuously(f"gs://{bucket}/match/*", interval=10.0, has_deduplication=True)
   | Map(lambda x: x.path)
   | ReadAllFromText()
   | Map(lambda x: logging.info(x))
)

它们在 Dataflow 中都可以正常工作。

【讨论】：