Error about mixing pipelines in nosetests
- Dominant language
- Java
- Stars
- 8.7k
- Forks
- 4.7k
- Avg merge
- 1d 20h
- Merged PRs (30d)
- 196
Description
When testing a PTranform (defined using @ptransform_fn) that merges several PCollections from different sources the following error is raised:
```
ValueError: Mixing value from different pipelines not allowed.
```
Actually running the same pipeline in GCP using the `DataflowRunner` does not give any error. Neither does running the test file manually instead of through nose.
Here is an example:
```
# Defined in module `utils`
@ptransform_fn
def Join(pcolls, by):
return pcolls | beam.CoGroupByKey()
```
```
class UtilsTest(unittest.TestCase):
def test_join(self):
p = TestPipeline(runner="DirectRunner")
p1 = (p
| "Create p1" >> beam.Create([
{'a': 1, 'b': 11},
{'a': 2, 'b': 22},
{'a': 3, 'b': 33}]))
p2 = (p
|
"Create p2" >> beam.Create([
{'a': 1, 'c': 111},
{'a': 1, 'c': 112},
{'a': 3, 'c': 333}]))
res = ((p1, p2) | "LeftJoin" >> utils.Join(by='a'))
beam.assert_that(res, beam.equal_to([
{'a': 1, 'b': 11, 'c': 111},
{'a':
1, 'b': 11, 'c': 112},
{'a': 2, 'b': 22},
{'a': 3, 'b': 33, 'c': 333}]))
# Run test pipeline
p.run()
```
Imported from Jira [BEAM-1996](https://issues.apache.org/jira/browse/BEAM-1996). Original Jira may contain additional context.
Reported by: while.
Contributor guide
Research direction
Start with the supplied UtilsTest.test_join reproduction, running it through nose and manually with DirectRunner to compare the behavior. Trace utils.Join, the @ptransform_fn wrapper, and beam.CoGroupByKey to identify why PCollections from the same pipeline appear mixed; done means the nose run no longer raises the cross-pipeline ValueError while the existing join assertions still pass.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- data-engineering, testing-qa
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 38/100