在Apache Beam中混合流式和非流式源时，转换节点AppliedTransform未按预期错误被DirectRunner替换

Question

当我声明一个有两个源（1 gcs和1 pubsub）的管道时，我得到一个错误，但仅限于Beam DirectRunner。有了谷歌数据流转轮，它工作得很好。我的管道有"流选项=True"

gcsEventsColl = p | "Read from GCS" >> beam.io.ReadFromText("gs://sample_events_for_beam/*.log") \
                  | 'convert to dict' >> beam.Map(lambda x: json.loads(x))
liveEventsColl = p | "Read from Pubsub" >> beam.io.ReadFromPubSub(topic="projects/axxxx/topics/input_topic") \
                   | 'convert to dict2' >> beam.Map(lambda x: json.loads(x))


input_rec = (gcsEventsColl, liveEventsColl) | 'flatten' >> beam.Flatten()

似乎DirectRunner为ReadFromText进行了一些不兼容的转换，但我不明白。

   File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/apache_beam/pipeline.py", line 564, in run
    return self.runner.run_pipeline(self, self._options)
   File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/apache_beam/runners/direct/direct_runner.py", line 131, in run_pipeline
    return runner.run_pipeline(pipeline, options)
   File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/apache_beam/runners/direct/direct_runner.py", line 529, in run_pipeline
    pipeline.replace_all(_get_transform_overrides(options))
   File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/apache_beam/pipeline.py", line 504, in replace_all
    self._check_replacement(override)
   File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/apache_beam/pipeline.py", line 478, in _check_replacement
    self.visit(ReplacementValidator())
   File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/apache_beam/pipeline.py", line 611, in visit
    self._root_transform().visit(visitor, self, visited)
   File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/apache_beam/pipeline.py", line 1195, in visit
    part.visit(visitor, pipeline, visited)
   File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/apache_beam/pipeline.py", line 1195, in visit
    part.visit(visitor, pipeline, visited)
   File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/apache_beam/pipeline.py", line 1195, in visit
    part.visit(visitor, pipeline, visited)   [Previous line repeated 4 more times]
   File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/apache_beam/pipeline.py", line 1198, in visit
    visitor.visit_transform(self)
   File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/apache_beam/pipeline.py", line 476, in visit_transform
    transform_node) RuntimeError: Transform node AppliedPTransform(Read from GCS/Read/SDFBoundedSourceReader/ParDo(SDFBoundedSourceDoFn)/ProcessKeyedElements/GroupByKey/GroupByKey,
    _GroupByKeyOnly) was not replaced as expected.

我想它与此代码有关，但我不确定：https://github.com/apache/beam/blob/master/sdks/python/apache_beam/runners/direct/direct_runner.py#L375

谢谢你的帮助

Answer 1

这是由于错误导致的内部故障。此错误消息表示Python DirectRunner在尝试重写转换时损坏了管道图。

Answer 2

事实上，相同的代码适用于Java direct runner。也许这是Python runner的一个限制？

  public static void main(String[] args) {
    Pipeline p = Pipeline.create();
    PCollection<String> apply = p.apply("read from gcs", TextIO.read().from("gs://xxx/*.log"));
    PCollection<String> apply1 = p.apply("read from pubsub", PubsubIO.readMessages().fromTopic("projects/xxx/topics/input_topic")).apply("test", MapElements.via(new FormatAsPubSubMessage()));
    PCollectionList<String> pcs = PCollectionList.of(apply).and(apply1);
    PCollection<String> merged = pcs.apply("merge", Flatten.<String>pCollections());
    merged.apply("log elements", MapElements.via(new LogElement()));
    p.run().waitUntilFinish();
  }