SparkForensics

Appearance

Apache DataFusion Comet ​

What it is ​

Apache DataFusion Comet is a high-performance accelerator for Apache Spark. It keeps Spark queries Arrow-native end to end: operators, expressions, shuffle and broadcast all stay in Apache Arrow columnar format[1]. Consecutive Rust-implemented operators are combined into a single block that is serialized as protobuf and executed by DataFusion on the executor[2]. The notes here follow Comet's documentation on its main branch, which is not versioned, so check them against the release you run.

Turning it on ​

The installation guide's quick start sets spark.plugins to org.apache.spark.CometPlugin, spark.shuffle.manager to org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager, spark.memory.offHeap.enabled to true with a spark.memory.offHeap.size, and spark.executor.memoryOverhead, and it turns on spark.comet.explain.fallback.enabled to log fallback reasons[3].

properties
spark.plugins=org.apache.spark.CometPlugin
spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager
spark.comet.explain.fallback.enabled=true
spark.memory.offHeap.enabled=true
spark.memory.offHeap.size=4g
spark.executor.memoryOverhead=2g

Comet requires JDK 17 or later, and the installation guide recommends using it only with Spark versions where both Comet and Spark tests run in CI[3:1].

Memory ​

Comet shares an off-heap memory pool with Spark, sized by spark.memory.offHeap.size. The pool is a shared budget rather than a shared allocator: Comet's native operators allocate from the Rust heap, but every reservation they make is charged against the same pool as Spark's own off-heap consumers. An operator that cannot spill fails the task when it cannot reserve the memory it needs[4].

Memory that Comet's accounting does not track has to fit in spark.executor.memoryOverhead, so enabling off-heap memory is not sufficient on its own. Setting spark.executor.memoryOverhead replaces the derived value (max(spark.executor.memoryOverheadFactor * spark.executor.memory, 384 MiB), factor 0.1) instead of adding to it, so a value below the derived one shrinks the container. The guide gives 2g as a reasonable start for a small executor, says to raise spark.executor.memoryOverheadFactor instead for large ones, and says to raise the overhead further if the cluster manager kills executors (on Kubernetes, ExecutorLostFailure with exit code 137)[4:1]. See Memory Management for how Spark sizes overhead.

Shuffle ​

Comet tries native shuffle first, then columnar shuffle, and falls back to Spark's shuffle if neither applies. spark.shuffle.manager is a static setting that cannot change at runtime, so Comet shuffle is toggled at runtime with spark.comet.shuffle.enabled. By default Comet converts a Spark ShuffleExchangeExec to columnar shuffle even when its child is a non-Comet plan, at the cost of a row-to-columnar conversion at the shuffle boundary, and spark.comet.shuffle.convertFromSparkPlan.enabled=false opts out. The codec is set with spark.comet.shuffle.compression.codec (lz4 by default, zstd and snappy also supported)[5]. The tuning rules on the Shuffle page describe Spark's own shuffle.

Joins ​

Vectorized engines tend to perform better with ShuffledHashJoin, so Comet's tuning guide suggests converting SortMergeJoin to it with spark.comet.exec.forceShuffledHashJoin=true. Comet does not yet provide spill-to-disk for ShuffledHashJoin, so this can run out of memory[6]. BroadcastHashJoinExec is supported, and BroadcastNestedLoopJoinExec falls back to Spark when the preserved side is broadcast[6:1]. See Join Optimization for the strategy rules on the Spark side.

UDFs ​

The presence of a Scala or Java UDF does not force the enclosing operator out of the Comet pipeline. Python @udf, pandas @pandas_udf, aggregate and Hive UDFs are out of scope and continue to fall back to Spark[7]. Comet can accelerate Python UDFs that use PyArrow-backed batch processing, such as mapInArrow and mapInPandas[8]. See PySpark Specifics for the UDF cost on the Spark side.

Scans ​

Comet falls back to Spark for Parquet scans it cannot convert. The documented cases include ShortType columns by default, ignoreMissingFiles or ignoreCorruptFiles set to true, and spark.sql.parquet.enableVectorizedReader=false[9]. Pushing filter evaluation into the Parquet decode step with spark.comet.parquet.rowFilterPushdown.enabled=true is disabled by default because it can hurt when the filter is not selective, and the scan tuning guide's workaround for scan tasks left with no row groups is a lower spark.sql.files.maxPartitionBytes[10]. See Data Formats for the Spark scan.

Reading the plan ​

Operators that Comet does not support remain in their original Spark form. Fallback can be partial (a subtree falls back to Spark while the rest stays in the Comet pipeline) or full (no Comet nodes appear). Wherever data crosses between columnar and row-based execution, Comet inserts a transition node such as CometColumnarToRow or CometSparkRowToColumnar, and frequent transitions usually indicate fallback inside the plan[2:1]. spark.comet.explain.fallback.enabled logs a warning with the reasons each query stage could not run in Comet[2:2]. Under AQE the report describes the plan before any adaptive re-planning, so coverage of the plan that finally executes can differ[2:3], which affects how you read it next to Adaptive Query Execution.

Repeated columnar-to-row and row-to-columnar conversions can dominate stage runtime. spark.comet.exec.transitionRevert.enabled=true reverts the whole stage to Spark row execution once the number of columnar-to-row transitions exceeds spark.comet.exec.transitionRevert.maxTransitions (default 2)[11].

Compatibility ​

Comet aims for results consistent with the Spark version in use and documents the areas where its behavior differs. Some expressions also have a native implementation that can differ from Spark for certain inputs, and these are not used unless you opt in with the relevant spark.comet.expression.<Name>.allowIncompatible=true setting[12].

Confidence ​

The settings, defaults and fallback cases above come from Comet's own installation guide, tuning guides and compatibility pages. Settings and their defaults can change between Comet releases, so confirm one against the release you run. The memory-sizing guidance is specific to Comet's off-heap accounting and does not replace sizing from your own executor metrics.

Limitations / false-positive risk ​

Fallback is per operator or subtree, so a plan with some Comet nodes can still run stages on the Spark engine. Because the fallback report under AQE describes the plan before adaptive re-planning[2:4], coverage in the final plan can differ from what the log lists.

Sources ​

  1. Apache DataFusion Comet README ↩︎

  2. Understanding Comet plans ↩︎ ↩︎ ↩︎ ↩︎ ↩︎

  3. Comet installation guide ↩︎ ↩︎

  4. Comet memory tuning ↩︎ ↩︎

  5. Comet shuffle tuning ↩︎

  6. Comet operator tuning ↩︎ ↩︎

  7. Comet Scala and Java UDFs ↩︎

  8. Comet PyArrow UDFs ↩︎

  9. Comet scan compatibility ↩︎

  10. Comet scan tuning ↩︎

  11. Comet transition tuning ↩︎

  12. Comet compatibility guide ↩︎