sarutak commented on PR #59551: URL: https://github.com/apache/spark/pull/59551#issuecomment-6224104031
Two more points, on the cost of the copy and on memory. **The benchmark of SPARK-56350** I ran `bench_arrow_columnar_udf.py` (one partition, 100-character strings in `data`). On master without this PR, and on v4.2.0, its Arrow columnar case fails with the same `IndexOutOfBoundsException` for 10000, 500000 and 2000000 rows. It worked when SPARK-56350 was merged (2d5c430), and it stopped working with SPARK-55715 (f8f4813, five days later; both are in 4.2.0): I ran it on f8f4813 and on its parent. Before SPARK-55715, `DataSourceRDD` closed the last reader of a task only at the end of the task; since then, `PartitionIterator.hasNext` closes a reader as soon as it is exhausted, before the last results come back from Python. The source of the benchmark frees its vectors only in `close()`, so it became one more case of the bug that this PR fixes. A source that refills or closes its vectors for every batch, such as `ArrowCachedBatchSerializer`, fails whatever `DataSourceRDD` does. The existing tests of `ArrowColumnarPythonUDFSuite` did not catch this, since their UDFs on `id` keep the cast and take the third path. With this PR, it runs again. Three rounds with `--rows 2000000` on a laptop (so only the ratio is meaningful): - columnar input enabled: 2697, 2643 and 2654 ms on average; - disabled (`ColumnarToRow`): 2887, 2889 and 2899 ms; - so 1.07x to 1.09x faster. SPARK-56350 reported 1.10x for 2M rows of 200-character strings and 1.17x for 10M rows of 100-character strings. With timers added around `copyColumns`, the copy took about 62 ms of a run of about 2.7 s (1954 batches, peak of the allocator 11.6 to 12.1 MB), so it takes a part of the gain but does not remove it. Could you add these numbers, or your own, to the description? **Memory** The copies are allocated from a child of `ArrowUtils.rootAllocator` with no limit, so they are not counted by the `TaskMe\ moryManager` and cannot spill, while the row queue path keeps the same rows in a `HybridRowQueue`, which is counted and can spill. How many batches are in flight is set by how far the input runs ahead of the results, not by the size of the pass-through columns. On the test source (4 narrow columns), the peak of the allocator was 5.9 MB for 200000 rows in batches of 100 and 20 MB for 2000000 rows in batches of 1000, so a few hundred batches were in flight, and this grows with the width of the pass-through columns. Before this PR, a source that allocates new vectors for every batch and does not free them early kept about the same memory alive, so this is not necessarily worse in practice, but it may be worth a sentence in the comment of `evalArrowColumnar`. -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
