damccorm commented on code in PR #41045:
URL: https://github.com/apache/beam/pull/41045#discussion_r4496144602
##########
sdks/python/apache_beam/runners/worker/operations.py:
##########
@@ -383,11 +383,14 @@ def receive_batch(self, windowed_batch):
"This is very inefficient, consider re-structuring your pipeline "
"or adding a DoFn to directly convert between these types.",
InefficientExecutionWarning)
+ # produce_batch takes a sequence, and some converters read it more
+ # than once (the pyarrow Table converter reads it once per column).
_cast_to_operation(consumer).process_batch(
windowed_batch.with_values(
consumer_batch_converter.produce_batch(
- self.producer_batch_converter.explode_batch(
- windowed_batch.values))))
+ list(
Review Comment:
I was nervous about this until I did more digging. For future readers, I
think that materializing `list(explode_batch(...))` is safe from a memory
standpoint because:
1) `windowed_batch.values` is already a single in-memory batch, and
`produce_batch` must return a single materialized in-memory batch (`pa.Table`,
`pd.DataFrame`, `np.ndarray`, etc.) for the downstream consumer anyway. So
we're going to do this sort of materialization immediately anyways.
2) As mentioned in the description, `BatchConverter.produce_batch(elements:
Sequence[E])` expects a `Sequence`, which all other call sites already pass.
Passing a list here also avoids issues when rebatching into converters like
`NumpyBatchConverter` or `PytorchBatchConverter` that reject generators. So I
think it is generically required, and we can't just push this into a consumer.
--
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]
