apache / apache/beam

AsIter side input is not correctly recognized as a dependency.

Open
#21,228 0 comments 0 reactions 0 assignees View on GitHub
bug core P3 python
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
2d 5h
Merged PRs (30d)
204

Description

The error is happening at current master (head). It is fine on the latest release (2.33.0).

Example to reproduce: 
```

import unittest

import apache_beam as beam

class LoggingFn(beam.DoFn):

def __init__(self,
name):
self._name = name

def process(self, element, *side_inputs):
print(f'Running {self._name}
(side inputs: {[list(s) for s in side_inputs]})')
return [self._name]

class BeamDagTest(unittest.TestCase):

def test_dag(self):
with beam.Pipeline() as p:
root = p | 'CreateRoot' >> beam.Create([None])

example_gen = root | 'CsvExampleGen' >> beam.ParDo(
LoggingFn('CsvExampleGen'),

)
statistics_gen = root | 'StatisticsGen' >> beam.ParDo(
LoggingFn('StatisticsGen'),

beam.pvalue.AsIter(example_gen), # AsIter to specify upstream task dependency.
)

schema_gen = root | 'SchemaGen' >> beam.ParDo(
LoggingFn('SchemaGen'),
beam.pvalue.AsIter(statistics_gen),

)
example_validator = root | 'ExampleValidator' >> beam.ParDo(
LoggingFn('ExampleValidator'),

beam.pvalue.AsIter(statistics_gen),
beam.pvalue.AsIter(schema_gen),
)

transform = root | 'Transform' >> beam.ParDo(
LoggingFn('Transform'),
beam.pvalue.AsIter(example_gen),

beam.pvalue.AsIter(schema_gen),
)
trainer = root | 'Trainer' >> beam.ParDo(

LoggingFn('Trainer'),
beam.pvalue.AsIter(example_gen),
beam.pvalue.AsIter(schema_gen),

beam.pvalue.AsIter(transform),
)
model_resolver = root | 'latest_blessed_model_resolver'
>> beam.ParDo(
LoggingFn('latest_blessed_model_resolver'),
)
evaluator = root
| 'Evaluator' >> beam.ParDo(
LoggingFn('Evaluator'),
beam.pvalue.AsIter(example_gen),

beam.pvalue.AsIter(trainer),
beam.pvalue.AsIter(model_resolver),
)

pusher = root | 'Pusher' >> beam.ParDo(
LoggingFn('Pusher'),
beam.pvalue.AsIter(trainer),

beam.pvalue.AsIter(evaluator),
)
```

 
According to AsIter [documentation](https://github.com/apache/beam/blob/64ec15fa2208d8f9b5ca5653866e1992fd07f7dc/sdks/python/apache_beam/pvalue.py#L527), entire PCollection should be made available as a side input, which means side input PTransform should run before the current PTransform. We used to exploit this feature to run DAG of tasks by injecting task dependency with side inputs, however this mechanism does not work properly in current master (71d7213d98):

Output with apache-beam\==2.33.0:
```

Running CsvExampleGen (side inputs: [])
Running latest_blessed_model_resolver (side inputs: [])
Running
StatisticsGen (side inputs: [['CsvExampleGen']])
Running SchemaGen (side inputs: [['StatisticsGen']])
Running
ExampleValidator (side inputs: [['StatisticsGen'], ['SchemaGen']])
Running Transform (side inputs:
[['CsvExampleGen'], ['SchemaGen']])
Running Trainer (side inputs: [['CsvExampleGen'], ['SchemaGen'],
['Transform']])
Running Evaluator (side inputs: [['CsvExampleGen'], ['Trainer'], ['latest_blessed_model_resolver']])
Running
Pusher (side inputs: [['Trainer'], ['Evaluator']])
```

Output with apache-beam installed from 71d7213d98 (origin/master):
```

Running CsvExampleGen (side inputs: [])
Running latest_blessed_model_resolver (side inputs: [])
Running
StatisticsGen (side inputs: [['CsvExampleGen']])
Running Pusher (side inputs: [[], []])
Running Evaluator
(side inputs: [['CsvExampleGen'], [], ['latest_blessed_model_resolver']])
Running SchemaGen (side inputs:
[['StatisticsGen']])
Running Trainer (side inputs: [['CsvExampleGen'], ['SchemaGen'], []])
Running
ExampleValidator (side inputs: [['StatisticsGen'], ['SchemaGen']])
Running Transform (side inputs:
[['CsvExampleGen'], ['SchemaGen']])
```

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

Contributor guide

Open the contributing guide

Research direction

Start with the Python SDK's AsIter documentation and implementation in apache_beam/pvalue.py, then reproduce the BeamDagTest pipeline shown in the issue on current master and release 2.33.0. Trace how ParDo side inputs contribute dependencies; done means the current master preserves the expected upstream execution order and the reproduced pipeline passes with the documented behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
data-engineering
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.