Accelerating Spark queries with Apache DataFusion Comet

• What problem is Comet solving?
• How is Comet put together?
• Which Spark versions does it support?
• How do you turn it on?
  • The memory setting is the one that bites
  • Proving it is on, from the plan
• An end-to-end example
  • Generating the data
  • The query
  • The same query, two plans
• What happens when Comet cannot run something?
• What does this cost?
• Recommendations
• Frequently asked questions
• References
• Trademarks

TL;DR

• Comet is a Spark plugin that rewrites the physical plan, replacing operators such as FileScan, Filter, HashAggregate and Exchange with native equivalents that run in Rust on Arrow arrays. The DataFrame and SQL you write does not change.
• Turning it on is three settings: a --packages coordinate, spark.plugins, and spark.shuffle.manager. The plugin sets spark.sql.extensions for you.
• It needs off-heap memory. Without spark.memory.offHeap.enabled=true (or spark.comet.exec.onHeap.enabled=true) the plugin logs one WARN and disables itself, and your query runs on stock Spark while looking like it worked.
• Read explain to confirm. A Comet prefix on the operator names is the only proof; the query returns the same answer either way.
• Fallback is per operator and it cascades upward. One unsupported expression can push its parent aggregate and the shuffle above it back onto the JVM.

What problem is Comet solving?

Spark’s SQL engine generates JVM bytecode and, for most operators, processes a row at a time. Whole-stage code generation removed much of the interpretation overhead, but the result is still JVM code walking Java objects, bounded by what the JIT will do and by garbage collection pressure on wide scans.

An analytical query spends most of its time in a small number of places: reading Parquet, filtering, hashing for joins and aggregations, and moving bytes for the shuffle. Those are exactly the operations that vectorized native code, working on columnar batches with SIMD, does considerably better than row-at-a-time JVM code.

Apache DataFusion Comet is a plugin that swaps those operators out. It is a subproject of Apache DataFusion, which supplies the native execution engine, and it uses Apache Arrow as the in-memory columnar format on both sides of the JNI boundary.

The design constraint that shapes everything else: Comet is not allowed to change your results or your API. It is a plan rewrite, not a new engine you port to. That is why it falls back rather than failing when it meets something it cannot do, and why the fallback behaviour is the part of this post worth reading twice.

How is Comet put together?

Three pieces cooperate, and knowing which one is which makes the configuration obvious rather than arbitrary.

flowchart TB
    subgraph JVM["Spark JVM"]
        SQL[DataFrame or SQL] --> CAT[Catalyst optimizer]
        CAT --> SPP[Spark physical plan]
        SPP --> RULE[CometExecRule<br/>plan rewrite]
        RULE --> MIXED[Plan with Comet operators]
    end
    subgraph NATIVE["Native, per executor"]
        DF[DataFusion execution]
    end
    MIXED -->|"Arrow batches over JNI"| DF
    DF -->|"Arrow batches back"| MIXED
    RULE -.->|"operator not supported"| FB[Left as a Spark operator]

• CometPlugin is a Spark plugin, named in spark.plugins. Its driver half registers the session extension; its executor half loads the native library.
• CometSparkSessionExtensions installs CometExecRule, which walks the physical plan after Catalyst is done and replaces supported operators. You do not set spark.sql.extensions yourself; the driver plugin does it, and says so:

INFO CometDriverPlugin: Setting spark.sql.extensions=org.apache.comet.CometSparkSessionExtensions

• CometShuffleManager replaces Spark’s shuffle manager so that shuffle writes and reads can stay columnar instead of round-tripping through rows.

The native library ships inside the jar, one per platform:

org/apache/comet/linux/aarch64/libcomet.so
org/apache/comet/linux/amd64/libcomet.so

That list is the whole platform story. The published jars carry Linux natives only, for amd64 and arm64. On macOS you build from source. The jar is large, around 88 MB, because those natives and the bundled Arrow JNI libraries are in it.

Which Spark versions does it support?

Comet publishes a separate artifact per Spark minor version and Scala binary version, so the coordinate encodes the compatibility:

There is no build for Spark 4.x on Scala 2.12, because Spark 4 is Scala 2.13 only. Spark 3.3 stopped receiving Comet releases at 0.7.0, and Spark 3.4 support is deprecated. The project’s own support table lists Java 17 for the 3.x line and Java 17 or 21 for the 4.x line.

Everything below was run on Spark 4.1.3, Scala 2.13, Java 21, with Comet 1.0.0.

How do you turn it on?

Three settings, plus a memory decision that is not optional.

spark-submit \
  --packages org.apache.datafusion:comet-spark-spark4.1_2.13:1.0.0 \
  --conf spark.plugins=org.apache.spark.CometPlugin \
  --conf spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager \
  --conf spark.memory.offHeap.enabled=true \
  --conf spark.memory.offHeap.size=4g \
  revenue_by_segment.py

The memory setting is the one that bites

Comet executes on Arrow buffers outside the Java heap, so it needs somewhere to put them. Leave the off-heap settings out and this is what happens:

WARN CometDriverPlugin: Comet plugin is disabled because Spark is not running in off-heap mode.

One WARN among thousands of INFO lines, and then the job runs to completion on stock Spark with correct results. Nothing fails. This is the single most common reason someone reports that Comet “made no difference”: it was never on.

If you cannot enable off-heap memory, Comet has a second mode:

  --conf spark.comet.exec.onHeap.enabled=true

Both work. With off-heap you get a different startup line, confirming Comet is sharing Spark’s off-heap pool rather than managing its own:

INFO CometDriverPlugin: Comet is running in unified memory mode and sharing off-heap memory with Spark
INFO core/src/lib.rs: Comet native library version 1.0.0 initialized

Treat that second line as the check that the native library actually loaded, because a plugin that initialises and a native library that loads are two different things.

Proving it is on, from the plan

Log lines scroll away. The durable check is the physical plan, where every accelerated operator carries a Comet prefix:

df.explain()

A one-operator smoke test is enough to see it:

CometHashAggregate [status#2, count#16L], [Final], [status#2], [count(1)]
+- CometExchange hashpartitioning(status#2, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=38]
   +- CometHashAggregate [status#2], [Partial], [status#2], [partial_count(1)]

If those names have no Comet prefix, Comet is not running, whatever the logs said.

An end-to-end example

Two tables, written as Parquet, then a query that exercises scan, filter, broadcast join, aggregation, exchange and sort.

Generating the data

Note the moduli. The status cycle has period 4, so deriving country from a period-4 expression would make status = 'delivered' select exactly one country and the demo would quietly measure nothing. Five countries and three segments are coprime with four, which keeps the filter and the grouping independent.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder.appName("comet-demo-data")
         .config("spark.ui.enabled", "false")
         .config("spark.eventLog.enabled", "false")
         .getOrCreate())

customers = (spark.range(0, 200_000)
             .select(
                 F.col("id").alias("customer_id"),
                 F.element_at(
                     F.array(F.lit("IN"), F.lit("US"), F.lit("DE"), F.lit("SG"), F.lit("BR")),
                     (F.col("id") % 5 + 1).cast("int")).alias("country"),
                 F.element_at(
                     F.array(F.lit("consumer"), F.lit("business"), F.lit("enterprise")),
                     (F.col("id") % 3 + 1).cast("int")).alias("segment"),
                 F.date_add(F.lit("2023-01-01").cast("date"),
                            (F.col("id") % 900).cast("int")).alias("signup_date"),
             ))
customers.write.mode("overwrite").parquet("/work/data/customers")

orders = (spark.range(0, 20_000_000)
          .select(
              F.col("id").alias("order_id"),
              (F.col("id") % 200_000).alias("customer_id"),
              F.element_at(
                  F.array(F.lit("placed"), F.lit("shipped"), F.lit("delivered"), F.lit("cancelled")),
                  (F.col("id") % 4 + 1).cast("int")).alias("status"),
              (F.col("id") % 997 + 1).cast("decimal(10,2)").alias("amount"),
              (F.col("id") % 7 + 1).cast("int").alias("quantity"),
              F.to_timestamp(F.date_add(F.lit("2025-01-01").cast("date"),
                                        (F.col("id") % 365).cast("int"))).alias("order_ts"),
          ))
orders.write.mode("overwrite").parquet("/work/data/orders")

print("customers:", spark.read.parquet("/work/data/customers").count())
print("orders   :", spark.read.parquet("/work/data/orders").count())
spark.stop()

That writes 20,000,000 order rows, about 179 MB of Parquet, and 200,000 customers.

The query

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder.appName("comet-demo-query")
         .config("spark.ui.enabled", "false")
         .config("spark.eventLog.enabled", "false")
         .config("spark.sql.shuffle.partitions", "8")
         .getOrCreate())

orders = spark.read.parquet("/work/data/orders")
customers = spark.read.parquet("/work/data/customers")

revenue = (orders
           .where((F.col("status") == "delivered") & (F.col("amount") > 100))
           .join(customers, "customer_id")
           .groupBy("country", "segment")
           .agg(F.sum("amount").alias("revenue"),
                F.count("*").alias("order_count"),
                F.avg("quantity").alias("avg_quantity"))
           .orderBy(F.col("revenue").desc()))

rows = revenue.collect()
print("=== RESULT ===")
for r in rows:
    print(f"{r['country']:<4} {r['segment']:<12} revenue={r['revenue']:>14} "
          f"orders={r['order_count']:>9} avg_qty={r['avg_quantity']:.3f}")
print("=== PHYSICAL PLAN ===")
print(revenue._jdf.queryExecution().executedPlan().toString())
spark.stop()

Fifteen groups, identical with Comet on or off:

SG   consumer     revenue=  164676437.00 orders=   299958 avg_qty=4.000
IN   business     revenue=  164676249.00 orders=   299958 avg_qty=4.000
BR   enterprise   revenue=  164675895.00 orders=   299958 avg_qty=4.000
US   consumer     revenue=  164675610.00 orders=   299957 avg_qty=4.000
DE   enterprise   revenue=  164674970.00 orders=   299956 avg_qty=4.000

The same query, two plans

Stock Spark 4.1.3:

*(4) Sort [revenue#12 DESC NULLS LAST], true, 0
+- Exchange rangepartitioning(revenue#12 DESC NULLS LAST, 8), ENSURE_REQUIREMENTS
   +- *(3) HashAggregate(keys=[country#7, segment#8], functions=[sum(amount#3), count(1), avg(quantity#4)])
      +- Exchange hashpartitioning(country#7, segment#8, 8), ENSURE_REQUIREMENTS
         +- *(2) HashAggregate(keys=[country#7, segment#8], functions=[partial_sum(amount#3), ...])
            +- *(2) Project [amount#3, quantity#4, country#7, segment#8]
               +- *(2) BroadcastHashJoin [customer_id#1L], [customer_id#6L], Inner, BuildRight, false
                  :- *(2) Project [customer_id#1L, amount#3, quantity#4]
                  :  +- *(2) Filter ((isnotnull(status#2) AND (status#2 = delivered)) AND (amount#3 > 100.00))
                  :     +- *(2) ColumnarToRow
                  :        +- FileScan parquet [customer_id#1L,status#2,amount#3,quantity#4]
                  +- BroadcastExchange HashedRelationBroadcastMode(...)

The same query with Comet:

*(1) CometColumnarToRow
+- CometSort [country#7, segment#8, revenue#12, order_count#13L, avg_quantity#14], [revenue#12 DESC NULLS LAST]
   +- CometExchange rangepartitioning(revenue#12 DESC NULLS LAST, 8), ENSURE_REQUIREMENTS, CometNativeShuffle
      +- CometHashAggregate [...], [Final], [country#7, segment#8], [sum(amount#3), count(1), avg(quantity#4)]
         +- CometExchange hashpartitioning(country#7, segment#8, 8), ENSURE_REQUIREMENTS, CometNativeShuffle
            +- CometHashAggregate [...], [Partial], [country#7, segment#8], [partial_sum(amount#3), ...]
               +- CometProject [amount#3, quantity#4, country#7, segment#8]
                  +- CometBroadcastHashJoin [customer_id#1L], [customer_id#6L], Inner, BuildRight
                     :- CometProject [customer_id#1L, amount#3, quantity#4]
                     :  +- CometFilter [...], ((isnotnull(status#2) AND (status#2 = delivered)) AND (amount#3 > 100.00))
                     :     +- CometNativeScan parquet [customer_id#1L,status#2,amount#3,quantity#4]
                     +- CometBroadcastExchange [customer_id#6L, country#7, segment#8]
                        +- CometFilter [customer_id#6L, country#7, segment#8], isnotnull(customer_id#6L)

Operator for operator:

The last row is the point. In the stock plan ColumnarToRow sits immediately above the scan, so everything from the filter upward is row-at-a-time. With Comet, the equivalent conversion has moved to the very top of the plan, and the whole pipeline underneath it stays columnar.

Note also that whole-stage codegen markers (*(2), *(3)) mostly disappear. Comet operators are not code-generated Java; they are native operators, so Spark’s codegen stages collapse to the few remaining JVM boundaries.

What happens when Comet cannot run something?

This is the section that decides whether Comet helps you or merely runs.

Comet’s rewrite is per operator. Anything it does not support is left as the Spark operator it already was, and the plan becomes a mix. Turn the reporting on and it tells you what it refused and why:

  --conf spark.comet.explain.fallback.enabled=true

Take the same join, but add a Python UDF that bands orders by risk:

from pyspark.sql.types import StringType

@F.udf(returnType=StringType())
def risk_band(country, amount):
    if country in ("IN", "BR") and amount > 800:
        return "review"
    return "clear"

flagged = (orders.where(F.col("status") == "delivered")
           .join(customers, "customer_id")
           .withColumn("band", risk_band(F.col("country"), F.col("amount")))
           .groupBy("country", "band").count()
           .orderBy("country", "band"))

Comet reports this:

WARN CometExecRule: Comet cannot execute some parts of this plan natively
   +-  HashAggregate [COMET: Comet aggregate that merges intermediate buffers requires a Comet child
                      aggregate when the intermediate buffer formats are incompatible with Spark.
                      Incompatible aggregate function(s): count]
       +-  BatchEvalPython [COMET: BatchEvalPython is not supported]
INFO CometExecRule: Reverting Comet columnar shuffle to Spark shuffle between HashAggregateExec and
                    HashAggregateExec (no Comet operator on either side to consume columnar output)

Read that from the bottom. The Python UDF is genuinely unsupported, which is no surprise: it is interpreted Python, and there is nothing to translate. But look at what it took with it. The HashAggregate above the UDF also fell back, because a native final aggregate cannot merge intermediate buffers produced by a JVM partial aggregate. And with JVM operators on both sides, the columnar shuffle between them had nothing left to do, so that reverted too.

The resulting plan:

*(4) CometColumnarToRow
+- CometSort [country#7, band#13, count#14L], [...]
   +- CometColumnarExchange rangepartitioning(...), CometColumnarShuffle
      +- *(3) HashAggregate(keys=[country#7, band#13], functions=[count(1)])
         +- Exchange hashpartitioning(country#7, band#13, 8), ENSURE_REQUIREMENTS
            +- *(2) HashAggregate(keys=[country#7, band#13], functions=[partial_count(1)])
               +- *(2) Project [country#7, pythonUDF0#26 AS band#13]
                  +- BatchEvalPython [risk_band(country#7, amount#3)#12], [pythonUDF0#26]
                     +- *(1) CometColumnarToRow
                        +- CometProject [amount#3, country#7]
                           +- CometBroadcastHashJoin [customer_id#1L], [customer_id#6L], Inner, BuildRight
                              :- CometProject [customer_id#1L, amount#3]
                              :  +- CometFilter [...], (isnotnull(status#2) AND (status#2 = delivered))
                              :     +- CometNativeScan parquet [customer_id#1L,status#2,amount#3]
                              +- CometBroadcastExchange [customer_id#6L, country#7]

The scan, filter and join are still native. But a CometColumnarToRow now sits in the middle of the plan, and everything from the UDF to the first exchange runs on the JVM. One unsupported expression cost far more than itself.

The practical rule: a fallback in the middle of a pipeline is much more expensive than a fallback at the edge, because it forces a columnar-to-row conversion and blocks native execution for every operator that depends on it. When a query underperforms with Comet enabled, this log is where to look first, and the fix is usually to replace one UDF with built-in expressions rather than to tune Comet.

Two shuffle implementations appear in these plans and they are not the same thing:

Seeing Exchange with no prefix in an otherwise Comet plan is a signal, not a detail.

What does this cost?

Comet is not free, and the trade-offs are structural rather than incidental.

Memory moves off the heap. Comet works on Arrow buffers outside the JVM heap, so executor sizing changes: you are now splitting a fixed container between heap and off-heap rather than giving it all to the JVM. An executor tuned for stock Spark is usually the wrong shape for Comet.

Another 88 MB on the classpath, per executor. That is native code plus Arrow JNI. It is a one-time distribution cost, but it is not nothing on a large or frequently-restarted cluster.

Coverage is partial, and silence is the failure mode. Unsupported operators do not error; they fall back. A query can be fully supported today and partly supported after someone adds a UDF, with no signal beyond the plan changing shape. If you depend on Comet for capacity, the plan is something to assert on in tests, not to check once.

Floating point and incompatible expressions. Comet exposes spark.comet.exec.strictFloatingPoint and a family of spark.comet.expression.<Name>.allowIncompatible switches precisely because a handful of operations have results that are not bit-identical to Spark’s. The defaults are conservative, which means some expressions fall back rather than risk differing. Loosening them is a correctness decision, so read what each one allows before turning it on.

Platform constraints are real. Published jars carry Linux natives only. A team developing on macOS either builds Comet from source or accepts that local runs do not exercise the accelerated path at all, which makes the plan assertion above more important rather than less.

On performance: the operators above genuinely do less work per row than their JVM equivalents, and that is the mechanism. How much it is worth on your workload depends on how much time your queries actually spend in scan, filter, aggregation and shuffle, versus in UDFs, wide row construction or writes. Measure it on your own data and hardware. Numbers from a container on a laptop, including the one this post was written on, would tell you nothing useful.

Recommendations

• Assert on the plan, not on the logs. A Comet prefix on the operator names is the only durable proof. Logs scroll; a plan assertion in a test does not.
• Set the off-heap configuration in the same commit as the plugin settings. Forgetting it is the most common way to run a whole benchmark against stock Spark by accident.
• Turn on spark.comet.explain.fallback.enabled while evaluating, and turn it off afterwards. It is verbose, it fires per plan rewrite, and its value is entirely in the first week.
• Hunt fallbacks in the middle of pipelines first. A UDF near the leaves that can be rewritten with built-in expressions is usually worth more than any configuration change.
• Re-check the plan after every query change. Coverage is per operator, so a one-line change to a transformation can move a query from fully native to half native.
• Size executors for two memory pools. Heap plus off-heap, not heap alone.

The mental model worth keeping: Comet does not make Spark faster. It replaces specific Spark operators with native ones, and everything it cannot replace stays exactly as slow as it was, plus a conversion. The value you get is therefore a property of your query shape, not of the plugin, and the physical plan is where that property is written down.

Frequently asked questions

Do I need to rewrite my queries to use Comet? No. Comet rewrites the physical plan after Catalyst has finished, so the DataFrame and SQL API you use is unchanged. The same query returns the same results with the plugin on or off; only the operator names in explain differ.

Why does Comet seem to do nothing after I enable it? Most often because off-heap memory is not enabled. The plugin logs Comet plugin is disabled because Spark is not running in off-heap mode and then stays out of the way, so the job succeeds with no Comet operators in the plan. Set spark.memory.offHeap.enabled=true with a size, or use spark.comet.exec.onHeap.enabled=true.

How do I confirm Comet is actually running? Call explain() and look for the Comet prefix on operator names. At startup you should also see Comet native library version ... initialized, which is a stronger signal than the plugin merely initialising.

Does Comet support Python UDFs? No. BatchEvalPython falls back to the JVM, and it typically drags the aggregate above it and the surrounding shuffle back as well. Replacing a UDF with built-in expressions is the highest-value change you can make to a Comet plan.

Does Comet work on macOS? Not from the published jars, which bundle Linux amd64 and arm64 natives only. macOS requires building from source.

Which Spark version should I use? Pick the artifact that matches your Spark minor version and Scala binary version, for example comet-spark-spark4.1_2.13 for Spark 4.1 on Scala 2.13. There is no Scala 2.12 build for Spark 4.x, and Spark 3.3 stopped receiving releases early.

Do I need to set spark.sql.extensions myself? No. The driver plugin sets it to org.apache.comet.CometSparkSessionExtensions during initialisation and logs that it did.

References

• Apache DataFusion Comet documentation for the installation guide, supported versions and the configuration reference
• Comet configuration settings for the full spark.comet.* list and each default
• Apache Arrow columnar format for the in-memory layout that crosses the JNI boundary
• CometExecRule, the rule that rewrites the physical plan and reports fallbacks
• Catalyst optimizer internals for the plan stages Comet rewrites after
• Spark memory management for the off-heap pool Comet shares in unified mode

Trademarks

Apache Spark, Apache DataFusion, Apache DataFusion Comet, Apache Arrow, Apache Parquet and Apache are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries.