在嵌入式 Flinkrunner (apache_beam [GCP]) 中使用 pub/sub io 运行光束流管道 (Python) 时出错

Question

我在 Flinkrunner 上的 Apache Beam 中运行流管道 (python) 时遇到以下错误。该管道包含 GCP pub/sub io 源和 pub/sub 目标。

WARNING:root:Make sure that locally built Python SDK docker image has Python 3.6 interpreter.

ERROR:root:java.lang.IllegalArgumentException: PCollectionNodes [PCollectionNode{id=ref_PCollection_PCollection_1, PCollection=unique_name: "23 Read from Pub/Sub/Read.None"
coder_id: "ref_Coder_BytesCoder_1"
is_bounded: UNBOUNDED
windowing_strategy_id: "ref_Windowing_Windowing_1"
}] were consumed but never produced
Traceback (most recent call last):
  File "<stdin>", line 5, in <module>
  File "/usr/local/lib64/python3.6/site-packages/apache_beam/pipeline.py", line 586, in __exit__
    self.result.wait_until_finish()
  File "/usr/local/lib64/python3.6/site-packages/apache_beam/runners/portability/portable_runner.py", line 599, in wait_until_finish
    raise self._runtime_exception
RuntimeError: Pipeline BeamApp-swarna0kpaul-0712135603-763999c_45da372e-757d-4690-8e25-1a5ed0a5cc84 failed in state FAILED: java.lang.IllegalArgumentException: PCollectionNodes [PCollectionNode{id=ref_PCollection_PCollection_1, PCollection=unique_name: "23 Read from Pub/Sub/Read.None"
coder_id: "ref_Coder_BytesCoder_1"
is_bounded: UNBOUNDED
windowing_strategy_id: "ref_Windowing_Windowing_1"
}] were consumed but never produced

我正在尝试在 Python 中运行以下代码 我正在尝试使用我在 GCP 帐户中创建的 2 个发布/订阅主题（{输入主题}、{输出主题}）运行 主题采用以下格式 - 项目/{项目名称}/主题/{主题名称}

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
input_topic=<input topic>
output_topic=<output topic>
options = PipelineOptions(["--runner=FlinkRunner", "--checkpointing_interval=1000","--streaming"])
with beam.Pipeline(options=options ) as pipeline:
  input1 = pipeline | " Read from Pub/Sub" >> beam.io.ReadFromPubSub(topic=input_topic).with_output_types(bytes)
  output = (input1
            |beam.WindowInto(beam.transforms.window.FixedWindows(5))
            |"Write to Pub/Sub" >>beam.io.WriteToPubSub(topic=output_topic, with_attributes=False).with_input_types(bytes))

系统中提供以下软件版本

Python 3.6.8
apache_beam [gcp]==2.30.0
java -version
openjdk version "1.8.0_292"
OpenJDK Runtime Environment (build 1.8.0_292-b10)
OpenJDK 64-Bit Server VM (build 25.292-b10, mixed mode)

我尝试按照本页中的规范使用 flink 集群和便携式 flink 运行程序运行它，但遇到了相同的错误。

当我使用以下选项时，相同的代码运行良好

options = PipelineOptions(["--streaming"])

Answer 1

该apache_beam.io.ReadFromPubsub()转换仅适用于 DirectRunner 和 Dataflow Runner，但您可以尝试外部转换：apache_beam.io.external.gcp.pubsub.ReadFromPubSub请参阅： https: //github.com/apache/beam/blob/release-2.39.0/sdks/python/apache_beam/io /external/gcp/pubsub.py#L39