apache / apache/beam

Python Direct Runner doesn't support both streaming & non streaming sources

Open
#21,103 10 comments 6 reactions 0 assignees View on GitHub
bug core direct P2 python runners
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

Please see Stack Overflow discussion:

[https://stackoverflow.com/questions/68125864/transform-node-appliedptransform-was-not-replaced-as-expected-error-with-the-dir](https://stackoverflow.com/questions/68125864/transform-node-appliedptransform-was-not-replaced-as-expected-error-with-the-dir)

When I create a GCS source & a Pub Source and try to flatten both, there is an error because of some incompatible transformation done by the direct runner.

Code example:
```

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()

```

Error:
```

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.

```

The direct runner corrupts the pipeline when it rewrites the transforms.

 

Imported from Jira [BEAM-12586](https://issues.apache.org/jira/browse/BEAM-12586). Original Jira may contain additional context.
Reported by: rodriguezc.

Contributor guide

Open the contributing guide

Research direction

Reproduce the mixed GCS ReadFromText and Pub/Sub ReadFromPubSub pipeline with Flatten, then inspect the replacement path in apache_beam/runners/direct/direct_runner.py and validation in apache_beam/pipeline.py. Done means the Direct Runner handles this pipeline without the AppliedPTransform replacement error and includes coverage for the case.

Written by the indexing model from the issue text.

Assessment

Tech stack
gcp, python
Domain
stream-processing
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.