jk-shah opened a new issue, #41042:
URL: https://github.com/apache/beam/issues/41042
### What happened?
A pipeline that reads a file and writes it straight to Parquet fails when the
data has more than one column:
```yaml
# pipeline.yaml
pipeline:
type: chain
transforms:
- type: ReadFromCsv
config: {path: users.csv}
- type: WriteToParquet
config: {path: out/users.parquet}
```
```
$ printf 'id,name\n1,Ada\n2,Grace\n' > users.csv
$ python -m apache_beam.yaml.main --yaml_pipeline_file=pipeline.yaml
...
pyarrow.lib.ArrowInvalid: Column 1 named name expected length 2 but got
length 0
[while running 'ReadFromCsv/Unbatch ...']
```
The same happens without YAML:
```python
import apache_beam as beam
with beam.Pipeline() as p:
_ = (
p
| beam.io.ReadFromCsv('users.csv')
| beam.io.WriteToParquet('out/users', schema=None))
```
Observed with Beam 2.75.0 (pyarrow 23.0.1, pandas 2.3.3) on PrismRunner,
FnApiRunner and DirectRunner. The code involved is unchanged on `master`.
* Single-column data works.
* `ReadFromCsv` -> `WriteToJson` works.
* `ReadFromJson` -> `WriteToParquet` fails the same way.
* Adding any step in between (for example a `Filter` or `MapToFields`) avoids
the failure, which is why it is easy to miss.
### Cause
`ReadFromCsv` produces pandas DataFrame batches and `WriteToParquet` consumes
`pyarrow.Table` batches, so the runner rebatches between them in
`GeneralPurposeConsumerSet.receive_batch`
(`sdks/python/apache_beam/runners/worker/operations.py`):
```python
consumer_batch_converter.produce_batch(
self.producer_batch_converter.explode_batch(windowed_batch.values))
```
`explode_batch` returns a generator, but `BatchConverter.produce_batch` is
declared to take a `Sequence[E]`. `PyarrowBatchConverter.produce_batch`
iterates its input once per column, so the first column consumes the
generator and every later column is empty.
### 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]
