You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "francisduffy (via GitHub)" <gi...@apache.org> on 2023/04/10 11:05:34 UTC
[GitHub] [beam] francisduffy opened a new issue, #26190: [Task]: Beam Python Version > 2.38.0 DirectRunner ~ AssertionError: A total of N watermark-pending bundles did not execute
francisduffy opened a new issue, #26190:
URL: https://github.com/apache/beam/issues/26190
### What needs to happen?
The minimal working example below works (with `DirectRunner`) with Python 3.9 and Apache Beam 2.38.0 but fails on Apache Beam 2.39.0 and 2.44.0.
Using Apache Beam 2.39.0 and 2.44.0, the example fails with the error `AssertionError: A total of 2 watermark-pending bundles did not execute.`. When I switch the logging to `DEBUG`, I see messages of the form `Unable to add bundle for stage` along with `Stage input watermark: Timestamp(-9223372036854.775000)` (i.e. `timestamp.MIN_TIMESTAMP`) and `Bundle schedule watermark: Timestamp(9223372036854.775000)` (i.e. `timestamp.MAX_TIMESTAMP`) for the two bundles.
I have marked this issue as a task because I am not sure if this change in behaviour is by design in version 2.39.0 or if it is actually an issue / bug.
Note: I have opened a question [here](https://stackoverflow.com/q/75961407) on this also.
```py
import logging
import apache_beam as beam
def setup_logging():
log_format = '[%(asctime)-15s] [%(name)s] [%(levelname)s]: %(message)s'
logging.basicConfig(format=log_format, level=logging.INFO)
logging.info("Pipeline Started")
class CreateKvPCollectWithSideInputDoFn(beam.DoFn):
def __init__(self):
super().__init__()
def process(self, element, side_input):
print(f"side_input_type: {type(side_input)}")
yield "b", "2"
class CreateKvPCollectDoFn(beam.DoFn):
def __init__(self):
super().__init__()
def process(self, element):
yield "a", "1"
def main():
setup_logging()
pipeline = beam.Pipeline()
pcollect_input = (
pipeline
| "Input/Create" >> beam.Create(["input"])
)
kvpcollect_1 = (
pcollect_input | "PCollection_1" >> beam.ParDo(CreateKvPCollectDoFn())
)
beamdict_1 = beam.pvalue.AsDict(kvpcollect_1)
kvpcollect_2 = (
pcollect_input
| "PCollection_2" >> beam.ParDo(
CreateKvPCollectWithSideInputDoFn(), side_input=beamdict_1
)
# Commenting in this line allows the example to work on 2.39.0
# | "Reshuffle_2" >> beam.Reshuffle()
)
kvpcollect_3 = (
(kvpcollect_1, kvpcollect_2)
| "Flatten" >> beam.Flatten()
)
beamdict_3 = beam.pvalue.AsDict(kvpcollect_3)
(
pcollect_input
| "UseBeamDict_3" >> beam.ParDo(CreateKvPCollectWithSideInputDoFn(), side_input=beamdict_3)
| "PrintResult" >> beam.Map(print)
)
result = pipeline.run()
result.wait_until_finish()
if __name__ == '__main__':
main()
```
### Issue Priority
Priority: 2 (default / most normal work should be filed as P2)
### Issue Components
- [X] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Samza Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on issue #26190: [Task]: Beam Python Version > 2.38.0 DirectRunner ~ AssertionError: A total of N watermark-pending bundles did not execute
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on issue #26190:
URL: https://github.com/apache/beam/issues/26190#issuecomment-1514542379
Thanks. This commit also caused other regressions in direct runner, see: https://github.com/apache/beam/issues/25315
The original author who was taking a look is currently OOO for a bit, so it may take some time to investigate and fix. But if you or someone else is willing to take a closer look, any help is appreciated.
In the meantime I'll add this to the backlog of Direct runner issues I am aware of for future prioritization.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] francisduffy commented on issue #26190: [Task]: Beam Python Version > 2.38.0 DirectRunner ~ AssertionError: A total of N watermark-pending bundles did not execute
Posted by "francisduffy (via GitHub)" <gi...@apache.org>.
francisduffy commented on issue #26190:
URL: https://github.com/apache/beam/issues/26190#issuecomment-1513829809
Thanks a lot for the tips on tracking down the offending commit. I believe that, bisecting between `v2.38.0` and `v2.39.0`, that the offending commit is 8a7c92cce1e6e059816ef9f667c96633f830f046 from the pull request #16841.
> Quick question: does your experiment use direct runner in Streaming or Batch mode ?
The experiment uses Batch mode.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on issue #26190: [Task]: Beam Python Version > 2.38.0 DirectRunner ~ AssertionError: A total of N watermark-pending bundles did not execute
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on issue #26190:
URL: https://github.com/apache/beam/issues/26190#issuecomment-1513276369
Thanks for opening! Note that you could probably find the offending commit with `git bisect`, if you are comfortable with running beam from master (s.apache.org/beam-python-dev-wiki has some tips on that)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on issue #26190: [Task]: Beam Python Version > 2.38.0 DirectRunner ~ AssertionError: A total of N watermark-pending bundles did not execute
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on issue #26190:
URL: https://github.com/apache/beam/issues/26190#issuecomment-1513277168
Quick question: does your experiment use direct runner in Streaming or Batch mode ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org