jk-shah opened a new issue, #41078:
URL: https://github.com/apache/beam/issues/41078
### What happened?
When running a Python or Beam YAML pipeline with the default `DirectRunner`
(`SwitchingDirectRunner`), if `PrismRunner` is bypassed (for example when the
pipeline contains an external cross-language transform per
`direct_runner.py:141`) or cannot start (offline environment / restricted
container where the Prism binary cannot be downloaded or executed),
`SwitchingDirectRunner` falls back to `BundleBasedDirectRunner`.
If the pipeline includes `WriteToParquet` (or any transform using
`DoFn.process_batch`), execution then fails with `NotImplementedError`, even
though the in-process `FnApiRunner` executes the exact same pipeline cleanly.
Minimal reproducer (forcing the non-Prism fallback path):
```python
import tempfile, apache_beam as beam, pyarrow as pa
schema = pa.schema([('id', pa.int64()), ('name', pa.string())])
with tempfile.TemporaryDirectory() as d:
with beam.Pipeline(runner='BundleBasedDirectRunner') as p:
(p
| beam.Create([{'id': 1, 'name': 'Ada'}])
| beam.io.WriteToParquet(f'{d}/out', schema=schema))
```
Cause
In sdks/python/apache_beam/runners/direct/direct_runner.py (lines 141–146
and 204–230), SwitchingDirectRunner unconditionally falls back to
BundleBasedDirectRunner(), which does not implement process_batch for batched
PyArrow DoFns (FnApiRunner does). Checking for process_batch transforms (or
routing batch pipelines without TestStream/unbounded state to FnApiRunner when
Prism is unavailable) avoids the failure.
### Issue Priority
Priority: 2 (default / most bugs 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 YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Prism 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
