Spark UDF Performance: Migrating UDF Jobs from EMR to Quanton with 40% Fewer Nodes

Spark UDF performance often depends more on the work around the UDF than on the UDF itself. Quanton keeps the UDFs in the JVM, as Amazon EMR does, and makes that surrounding work cheaper. A B2B FinOps vendor measured the result on the UDF pipeline it runs for each of its customers: 34.4% less runtime than EMR for its p99 tenant, and for its largest tenant Quanton finished within 3% of EMR’s runtime on 40% fewer nodes.

Many production Spark jobs are pipelines built around user-defined functions, with Scala UDFs and Row lambdas in the middle of the plan, rows that are hundreds of columns wide, repartition and persist calls, and skewed partitioned writes to object storage. These jobs are slow on both kinds of Spark engine. A JVM engine such as EMR runs every part of the job in the JVM, so on rows that wide most of the CPU goes to copying rows between operators, and the shuffle and spill data fill local disks. A native engine such as Apache Gluten or Apache DataFusion Comet cannot run a UDF natively, so it converts columnar batches to rows before each UDF and back after it, and on a wide schema those conversions can cost more than the native operators save.

Results at a glance

Tenant	Processed rows	EMR	Quanton	Result
p50	8.0M	96 s (28 nodes)	83 s (28 nodes)	13.5% less runtime
p97	939M	4,196 s (28 nodes)	3,095 s (28 nodes)	26.2% less runtime
p99	1.83B	9,158 s (26 nodes)	6,091 s (26 nodes)	34.4% less runtime
p100 (largest)	19.5B	Did not finish; ran out of local disk and aborted after 3.7 hours (180 nodes)	7,281 s (180 nodes)	Quanton finished on the same nodes, EMR did not
p100 (largest)	19.5B	7,092 s (300 nodes)	7,281 s (180 nodes)	Within 3% of EMR’s runtime with 40% fewer nodes and 38.4% fewer core-hours

The customer ran every comparison on its own AWS account, with m8gd nodes, the same input, and the same output on both engines. For p50, the times exclude executor startup and the setup code that both engines run before the first query.

EMR also failed twice on clusters where Quanton finished. On the largest tenant, EMR filled the local disks of all 180 nodes with shuffle and spill data, and the customer aborted the run after 3.7 hours with no output, so all of that compute was wasted. EMR needed 300 nodes to finish. Quanton wrote 67% less data to local disk, because its engine architecture keeps shuffle and spill data columnar and compact, so the same job finished on 180 nodes.

The customer’s job

The customer is a B2B SaaS company, a FinOps vendor that does cost optimization for infrastructure. Like many B2B SaaS companies, it processes data on behalf of its customers, and it runs its cost-tracking pipeline separately for each customer, or tenant. The pipeline is a Scala Spark job built around UDFs, and its plan has every part that makes a UDF job hard to accelerate:

• A chain of Scala UDFs and Row lambdas over a schema of about 500 columns.
• A join followed by explode, which more than doubles the row count.
• A wide round-robin repartition and a persist(DISK_ONLY) of the enriched data.
• An hourly write and a daily aggregate with a write, both partitioned and both to S3.

B2B customers come in very different sizes, and this customer’s tenants follow a real distribution that covers several orders of magnitude. The median tenant processes 8 million rows and the largest processes 19.5 billion, so the same job has to run well at both ends. The customer compared the engines at several tenant-size percentiles, from the p30 tenant up to the p100 tenant. The success criteria were cost, scaling, and concurrency on the deployment model it planned to run in production, which was Quanton on its own Amazon EKS cluster through the Quanton Kubernetes Operator.

Why Spark UDF jobs are slow on JVM and native engines

A UDF job like this customer’s is expensive on both kinds of Spark engine, for different reasons. Both also pay for persist of wide data and for partitioned writes, which add a sort, an aggregation, and a commit for each task. TPC-DS barely touches the UDF, cache, and partitioned-write paths, so a good TPC-DS result says little about how such a job will run.

On a JVM engine such as Amazon EMR

EMR runs every operator in the JVM, so the UDFs need no conversion, and the cost sits in the rest of the plan. Spark copies a wide row once for each operator in a chain of JVM operators, and on a row of 500 columns each copy moves a lot of data. The UDF itself is often a small share of the CPU, and these copies are the large share. Spark’s shuffle and spill data are row-oriented, so a wide schema takes a lot of local disk. On the customer’s largest tenant, EMR wrote 280,892 GB of shuffle and spill data on its 300-node run, and on 180 nodes it ran out of disk.

On a native engine such as Gluten or Comet

Gluten with Velox and Comet speed up Spark SQL and DataFrame jobs built from built-in expressions. They run the scan, filters, joins, aggregations, and shuffle as native vectorized code over columnar batches, and finish TPC-DS several times faster than the JVM. Most published Spark benchmarks, ours included, measure that case.

A native engine runs the operators it has a native version of and hands the rest back to Spark’s JVM execution. A Scala UDF, a typed map over Row objects, or a custom Row decorator has no native version, so the engine falls back to the JVM for that operator. To do that, the engine converts columnar batches to rows before the operator and converts the rows back to columns after it. One conversion is cheap. A Spark job that calls UDFs in every stage pays for a conversion in every stage, and on a row of 500 columns each conversion copies a lot of data.

Wide rows also change the cost of the shuffle. A wide UDF job often repartitions into tens of thousands of partitions, so each shuffle block between one map task and one reduce task holds only a few rows. A JVM engine reads those blocks as one stream of rows. A columnar engine that turns every block into its own small batch pays a fixed cost per batch, per column, in every operator after the shuffle. On a columnar plan, persist also encodes and decodes the cached data in the JVM.

Native engines can even make a UDF job slower. The native engine takes a large share of executor memory off-heap, and the UDFs and row operators that still run in the JVM get a smaller heap, so they spill and garbage collect more often.

How Quanton accelerates Spark UDF jobs

Quanton runs every operator that has a native version as native code and keeps the UDFs in the JVM. It then removes the costs described above on both sides: the row copies and large shuffle and spill data of JVM execution, and the conversions and small batches of native execution. We did not ask the customer to tune its job or change its code. All of the gains came from changes in the engine, and they fall into three groups.

1. Keep data columnar and in full-size batches across wide shuffles

After a wide shuffle, Quanton merges the small blocks that arrive at a reducer into full-size batches, so the operators after the shuffle work on thousands of rows at a time instead of a handful. Before a wide shuffle, it merges small partitions on the map side, so the shuffle has fewer, larger outputs. For wide schemas, it picks a shuffle layout that stays compact when the number of partitions times the number of columns is large. It also distributes a round-robin repartition in a way that is deterministic on retry without an extra sort of the whole dataset.

2. Cut the cost of the JVM operators around each UDF

The UDFs still run in the JVM, as they do on EMR. What Quanton changes is the work around them. Where Spark would copy a wide row once for each operator in a chain of JVM operators, Quanton fuses the chain so the row is copied once. In the customer’s enrichment stage, most of the CPU went to those copies and very little went to the UDF bodies, so this change removed most of the stage’s overhead.

3. Make the cache and the write phase cheaper

Quanton stores persist data as compressed columnar pages and reads it back without a JVM decode, so the cached data is smaller on disk and faster to read. It stops a partial aggregation early when the aggregation is not reducing rows. It skips the commit work for write tasks that wrote no rows, which matters for a partitioned write where most tasks have nothing to write.

All of these changes apply to any Spark job that combines UDFs, wide rows, many shuffle partitions, persist, and partitioned writes.

What the customer measured against EMR

The customer ran each comparison against its tuned EMR setup, with the same input and the same output on both engines.

On the p99 tenant, Quanton’s lead over EMR grew from 27.6% on the earlier release to 34.4% with the changes above:

Engine	p99 tenant runtime	vs EMR
EMR 7.6	9,158 s
Quanton v0.49.0	6,632 s	27.6% less runtime
Quanton with the UDF-job accelerations	6,091 s	34.4% less runtime

The p100 tenant is the customer’s largest, with 7.45 billion input rows that expand to 19.5 billion rows during the job. All runs used m8gd.4xlarge nodes with one executor per node:

Run	Nodes	Runtime	Core-hours
Quanton	180	7,281 s	5,097
EMR	180	Did not complete; aborted after 3.7 hours
EMR	300	7,092 s	8,274

Each executor had 46 GB of memory on both engines. The shuffle and disk numbers show why EMR needed more nodes:

Metric	EMR, 300 nodes	Quanton, 180 nodes	Difference
Shuffle written	108,336 GB	54,728 GB	49.5% less
Disk spilled	172,556 GB	37,896 GB	78.0% less
Total local disk written	280,892 GB	92,624 GB	67.0% less
Local disk written per node	936 GB	515 GB	45.0% less
Output Parquet files	57,486	57,486	Same

Each m8gd.4xlarge node has about 855 GB of usable local NVMe disk. On 180 nodes, EMR’s local disk use reached 95.2% on the busiest node, and YARN marked all 180 nodes unhealthy for about 50 minutes after disk use crossed its 90% threshold. The 3.7 hours that 180 nodes spent on the failed run produced no output. EMR needed 300 nodes to finish. Quanton finished on 180 nodes within 3% of EMR’s 300-node runtime, with 40% fewer nodes, 38.4% fewer core-hours, and 67% less total local disk.

The same thing happened earlier on the p99 tenant. On the customer’s first 28-node cluster, YARN marked most EMR nodes unhealthy after local disk use passed 90%, and the run failed. Quanton finished on the same nodes.

After these runs, the customer migrated the pipeline from EMR to Quanton on its own EKS cluster.

Settings the customer used

The accelerations above need no configuration. The customer ran Quanton on its own Amazon EKS cluster through the Quanton Kubernetes Operator, with this setup:

• Nodes. The customer used m8gd.4xlarge nodes, with 16 vCPUs and about 855 GB of usable local NVMe disk each. Shuffle and spill go to the local NVMe disks, so disk capacity per node decides how large a job a cluster can finish.
• One executor per node. Each node ran one executor with 14 cores and 46 GB of memory. Fewer, larger executors mean fewer pods, which keeps the number of pod IP addresses low on Amazon EKS.
• Pod IP addresses. With the default Amazon VPC CNI settings, each node held about 60 IP addresses. The customer set WARM_IP_TARGET=2 and MINIMUM_IP_TARGET=3, which brought this down to about 4 IP addresses per node with no change in runtime. This matters at p100 scale, where the cluster has hundreds of nodes in a shared VPC.
• Static executors. The customer used a fixed number of executors instead of dynamic allocation, so both engines got the same resources in every run.
• Native memory. The customer set spark.quanton.executor.offHeapRatio=0.5, which gives the native engine half of executor memory, so the native sort before a partitioned write spills less.

Should you migrate your Spark UDF jobs from EMR to Quanton?

Yes. Quanton runs every operator that has a native version natively, keeps the UDFs in the JVM, and makes the work around each UDF cheaper: the row and columnar conversions, the shuffle, the cache, and the write. For this customer, Quanton ran the same job faster than EMR at every tenant size from p50 to p99, and on the largest tenant it finished within 3% of EMR’s runtime on 40% fewer nodes.

Quanton runs on Amazon EKS, including EMR on EKS as a custom image, and existing Spark jobs run on it with no code changes. For Spark SQL and DataFrame jobs, see our comparison of Quanton and the latest EMR. To find out which of your jobs would gain the most, run the free Spark cost analyzer on your EMR event logs, then run your largest UDF job on Quanton in your own VPC and compare the two event logs.

Frequently asked questions

How do I improve Spark UDF performance?

Reduce the work around the UDF, which often costs more than the UDF itself. That work is converting columnar data to rows and back, copying wide rows between JVM operators, processing small batches after wide shuffles, and caching and writing wide data. Replace a UDF with built-in expressions where you can, so a native engine can run that part of the plan. For the UDFs you keep, run the job on an engine that makes the surrounding work cheaper. For this customer’s p99 tenant, the UDF job ran with 34.4% less runtime on Quanton than on EMR, with no code changes.

Why are Spark UDF jobs slow on Amazon EMR?

EMR runs the whole job in the JVM. On a wide schema, Spark copies each row once for each operator in a chain of JVM operators, and its row-oriented shuffle and spill data take a lot of local disk. On this customer’s largest tenant, EMR needed 300 nodes to finish a job that Quanton finished on 180.

Why doesn’t native execution speed up Spark UDF jobs?

Native engines such as Apache Gluten and Apache DataFusion Comet run only the operators they have a native version of. A Scala UDF or a Row lambda runs in the JVM, so the engine converts data from columns to rows and back around it. On a wide schema with many such operators, the conversions and the small batches after wide shuffles can cost more than the native operators save.

How do I speed up Python UDFs in PySpark?

Use Apache Arrow, which Spark uses to transfer data efficiently between the JVM and Python processes. Pandas UDFs run with Arrow to transfer the data and pandas to work with it, which allows vectorized operations. Since Spark 4.2, Arrow optimization is also on by default for regular Python UDFs. The setting is spark.sql.execution.pythonUDF.arrow.enabled, and you can make a single function an Arrow Python UDF by passing useArrow=True to udf(). The measurements in this post come from a Scala UDF job.

Do I need to rewrite my UDFs to run them on Quanton?

No. Quanton runs the same Spark SQL, DataFrame, and PySpark APIs, and Scala UDFs run in the JVM as they do on EMR. Quanton makes the work around the UDFs cheaper, so the same job runs faster with no change to its code.

Why did EMR fail on 180 nodes when Quanton did not?

EMR wrote about 3 times as much shuffle and spill data to local disk as Quanton for the same job, 280,892 GB against 92,624 GB. On 180 nodes, local disk use passed YARN’s 90% health threshold, and YARN marked every node unhealthy. Quanton’s columnar shuffle and cache store a wide schema in less space, so the same job fit on the same 180 nodes.

Is Quanton faster than EMR for Spark UDF jobs?

Yes. Scala UDFs run in the JVM on Quanton, as they do on EMR, and Quanton accelerates the work around them, so a UDF job does not get slower when you move it from EMR to Quanton. In this customer’s tests, Quanton was faster than EMR at every tenant size from p50 to p99, and it finished the largest tenant on 180 nodes, where EMR ran out of local disk.