In several documented large cross-walk analyses on the KBase Data Lakehouse, failures happened when results computed across the Spark cluster were moved into a single driver process. The general warning is that .toPandas() pulls all data from the Spark cluster to the driver node. This is slow for large results and can cause out-of-memory errors, so filtering, joins and aggregations should be done in Spark first 1. The evidence comes from pitfalls met in individual analyses, not from systematic benchmarks. It does not show that collection is the main or dominant scaling limit in general, only that it was the failure point in the cases below. These cases fall into two kinds: a hard cap on serialized result size, and driver memory used up by pandas work after collection 1.
The hard ceiling on collected results
In one documented case, 1B-row tables were joined (kbase.genomes.feature × contig_x_feature × pangenome.gene_genecluster_junction) and filtered to >200K elements via broadcast. Calling .toPandas() on the Spark Connect result hit spark.driver.maxResultSize = 1024 MB and threw Total size of serialized results bigger than spark.driver.maxResultSize 1. The source states that the cap cannot be raised with SET spark.driver.maxResultSize at runtime, because the property is read-only after session start 1.
The source gives an estimate, not a measurement, of the scale at which the cap matters. A filtered contig_x_feature table of 218K bacteroidota contigs × ~140 features/contig gives 30M+ rows. At ~50 bytes serialized per row, that is ≈ 1.5 GB. The source calls this the standard pattern in any project doing genome-context cross-walks at >200 species 1.
The same ceiling appeared in a pangenome extraction. Gene-cluster memberships for klebsiella-pneumoniae (250 genomes × ~5,500 genes per genome) were extracted via gene JOIN gene_genecluster_junction WHERE genome_id IN (...), and the extraction exceeded Spark’s spark.driver.maxResultSize (1GB default) 1. The query succeeded in Spark and failed only when results were collected to the driver node. This supports, for this case, the observation that the failure happened at collection and not during distributed execution 1.
Driver memory, not just result size
Staying under the result cap is not enough, because pandas operations after collection can use up driver memory. One example is a spatial-range merge done in pandas: focal_features × contig_features on contig_id, then a filter by ±NEIGHBOR_BP. This builds a Cartesian product and then filters it. For 80K focal features × 21K contigs × ~105 features/contig on average, the merged DataFrame is ~24M rows × 9 columns. That is a ≈ 9 GB working set, and it ran out of memory on a 16 GB driver 1.
This failure came up in a polysaccharide utilization loci (PUL, clustered polysaccharide-degradation genes) gene-neighborhood analysis in Bacteroidota. The full run at 723K focal × 210K contigs failed. A run sampled down to 309 species still ran out of memory at 80K × 21K 1. By contrast, a photosystem ii (PSII) neighborhood at 27K focal × 16K contigs, with a 16M-row merge, fit in memory. The PUL and mycolic acid merges at full scale did not 1. This refines the scaling picture: in these analyses, success depended on the size of the intermediate merge, not on the inputs alone. The boundary is documented only for these few analyses 1.
A third case involved an exploded intermediate. Tree-based donor inference computed a per-(donor genus × KO) “donor candidate event” count by exploding 6.3M gain events × ~20 candidate-donor genera per event. The source puts this at about a 126M-row exploded DataFrame of ~19 GB. These sizes are the source’s approximate figures, not reported measurements. The step ran out of memory 1.
Remedies
The reported workaround for the result-size cap is staging. Write the Spark result to MinIO with df.coalesce(N).write.mode("overwrite").parquet("s3a://cdm-lake/..."), then read it back with spark.read.parquet(path).toPandas(). According to the source, the read-back uses Spark Connect’s parquet streaming path, which does not hit the result-size cap 1. The stated alternative is to read the staged data directly with pyarrow.parquet.read_table(path) via s3fs, provided PyArrow is configured. The source gives no timings for either route 1.
The source proposes two workarounds for the spatial-merge memory failure. The first is to process focal features in batches of ~10K, merge each batch and accumulate an aggregate result. The second is to do the spatial-range filter in Spark by pushing the BETWEEN clause through, so that only the per-feature aggregate is collected to the driver 1. Both are presented as proposals, and the source reports no outcome for either on the failed PUL run 1.
The donor-inference explode was replaced with an algebraic identity. Take each (family F × KO K) with R_FK total recipient events. Every genus G in F that has K present is a candidate donor for (R_FK − recipient_events_for_G). This can be computed as a single join plus a subtraction, with no explode. Stage 2b ran in 142s instead of running out of memory 1. This is the only remedy here with a reported execution time. The source does not compare it against running the original explode on a larger driver 1.
For pangenome extractions involving species with >200 genomes, the source recommends splitting the genome list into batches of 50-100 and concatenating the results. Its alternative is to increase spark.driver.maxResultSize in the Spark session config. It also notes that the per-species extraction script catches this error with try/except and continues to the next species 1.
Unresolved configuration question
The two entries on spark.driver.maxResultSize refer to different configuration stages, so they do not necessarily contradict each other. One says the property cannot be raised via SET at runtime because it is read-only after session start, and recommends Parquet staging 1. The other suggests raising it in the Spark session config as an alternative to batching 1. Neither entry says whether users of Spark Connect sessions on the Lakehouse can set the property when the session is created. That question remains open 1.
Open Directions
- Test whether
spark.driver.maxResultSizecan be set at Spark Connect session creation on the KBase Data Lakehouse. This would settle the configuration question above and show whether batching or staging is the only way past the cap 1. - Rerun the Bacteroidota PUL neighborhood with the BETWEEN filter pushed into Spark, collecting only per-feature aggregates. This would show whether push-down removes the driver-memory failure at full scale, where the source reports only the failure 1.
- Benchmark Parquet staging against direct
pyarrow.parquet.read_tablereads on the same 1B-row join cross-walk. This would measure wall-time and memory for both reported cap workarounds, which have no timings in the source 1. - Apply join-and-aggregate reformulations like the donor-inference identity to other explode-heavy steps, and compare them with larger-driver runs. This would test whether algebraic rewriting generalizes beyond the single Stage 2b case 1.
- Profile where wall-time and failures occur across a set of large analyses, covering distributed execution as well as driver collection. This would test whether driver collection is in fact the most common scaling limit, which the current single-case pitfalls cannot establish 1.
Related: join-strategy-and-table-partitioning-cost, vectorization-versus-row-wise-iteration, silent-failure-modes-in-distributed-queries, spark sql, pitfalls.