Crossing between Spark and pandas

Symptom

java.lang.OutOfMemoryError: Java heap space          (driver)
Total size of serialized results of 512 tasks (1024.0 MiB)
is bigger than spark.driver.maxResultSize (1024.0 MiB)

Or the notebook hangs for minutes and then the Python process is killed.

Why it happens

• toPandas() pulls every row to the driver, then builds a pandas DataFrame there. Data exists in the JVM, in transit and in pandas at the same moment, so peak memory is a multiple of the data size.
• It's often called "just to look" at a table, or on data that was small in development and grew in production.

How to confirm

1. The failure is in the driver log, right after the toPandas call.
2. Check the result size first: df.count() and the number of columns.

Fix

• Aggregate or filter in Spark first; convert only the small result.
• For a look at the data, use df.limit(1000).toPandas() or df.show().
• For handing data to another system, write it to a table or files instead.
• Enable Arrow (spark.sql.execution.arrow.pyspark.enabled=true). It cuts conversion time and overhead, but every row still lands on the driver.

Prevent

• Team rule: toPandas() only after an aggregation or a limit, never on raw tables.

Symptom

UserWarning: toPandas attempted Arrow optimization because
'spark.sql.execution.arrow.pyspark.enabled' is set to true; however,
failed by the reason below: ...
Attempting non-optimization as
'spark.sql.execution.arrow.pyspark.fallback.enabled' is set to true.

The conversion still works, just many times slower, and the warning is easy to miss in a job log.

Why it happens

• A column type Arrow can't convert in your Spark version (some nested or complex types), or a version mismatch between PySpark, PyArrow and pandas.
• Fallback is on by default, so Spark quietly switches to row-by-row conversion instead of failing.

How to confirm

1. Search driver logs for attempted Arrow optimization.
2. Re-run with fallback disabled to see the real error:

   spark.conf.set("spark.sql.execution.arrow.pyspark.fallback.enabled", "false")

Fix

• Cast or flatten the unsupported column (for example to_json on a complex struct) before converting.
• Align PyArrow and pandas with the versions your Spark release supports.

Prevent

• Disable fallback in tests and CI so incompatibilities fail loudly there instead of slowing production.

Symptom

No error; the job is just slow. The stage running the UDF dominates the runtime, executor CPU sits in Python processes, and the plan shows a BatchEvalPython node (plain Python UDF) or ArrowEvalPython (pandas UDF).

Why it happens

• A plain Python UDF processes one row at a time: each row is serialised in the JVM, sent to a Python worker, processed and sent back.
• Catalyst can't see inside the UDF, so filters that depend on its output can't be pushed down.
• Python workers also use memory outside the JVM heap, which can lead to "container killed" errors (see Module 1).

How to confirm

1. df.explain() shows BatchEvalPython.
2. Compare stage time with the UDF replaced by a no-op or built-in.

Fix

• Rewrite with built-in functions where possible: the biggest win.
• Otherwise use a pandas UDF: Arrow batches and vectorised pandas or NumPy code (why it's faster).
• On Spark 3.5+, an Arrow-optimised Python UDF (@udf(useArrow=True)) keeps row-wise code but speeds up data transfer.
• Apply UDFs after filtering, so fewer rows reach Python.

# before: row-at-a-time Python UDF
@udf("string")
def tier(amount): return "high" if amount > 1000 else "low"

# after: built-in, stays in the JVM and is optimised
df.withColumn("tier", F.when(F.col("amount") > 1000, "high").otherwise("low"))

Symptom

A pipeline that does df.rdd.map(…).toDF() in the middle is much slower than expected, reads far more data than its inputs suggest, and its plan shows full scans with no PushedFilters.

Why it happens

• Catalyst optimises up to the .rdd call and after the toDF(), but can't optimise across it. Filters later in the pipeline can't be pushed back into the scan.
• Rows are converted from Tungsten's binary format into Python Row objects and back.
• toDF() without a schema samples rows to infer types: an extra job, and sometimes the wrong types.

Fix

• Express the logic with built-ins, a pandas UDF or mapInPandas (see Examples).
• If an RDD step truly can't be avoided, filter and select columns before .rdd, and pass an explicit schema to createDataFrame.

Symptom

ValueError: Cannot combine the series or dataframe because it comes from
a different dataframe. In order to allow this operation, enable
'compute.ops_on_diff_frames' option.

Or code that was instant in pandas runs very slowly, or a step quietly moves all data into one partition.

Why it happens

• The index is expensive. pandas assumes a row order and index; Spark has neither natively. Creating a sequential default index, or aligning two DataFrames by index, needs extra passes or joins.
• Combining two different DataFrames requires a join on the index, which is disabled by default because it's costly.
• Some operations don't distribute well: anything needing a global row order, or positional access like iloc on big data.

Fix

• Set index_col when reading or converting, or use compute.default_index_type = "distributed" when consecutive index values don't matter.
• Join explicitly on a key column instead of relying on index alignment; enable compute.ops_on_diff_frames only when you accept the cost.
• Switch to the Spark DataFrame API (psdf.to_spark()) for heavy steps.

Symptom

[JVM_ATTRIBUTE_NOT_SUPPORTED] Attribute `sparkContext` is not supported
in Spark Connect as it depends on the JVM.

[NOT_IMPLEMENTED] rdd() is not implemented.

Exact messages vary by Spark and Databricks version.

Why it happens

With Spark Connect (Spark 3.4+), your Python code is a thin client that sends unresolved plans to a remote Spark server. There's no JVM in your process, so JVM-bound APIs such as sparkContext, RDDs and df.rdd aren't available. Databricks serverless and shared-access compute work this way.

Fix

• Rewrite RDD logic with DataFrame operations, built-ins, pandas UDFs or mapInPandas.
• Replace sc.broadcast with broadcast joins, sc.parallelize with spark.createDataFrame, and accumulators with aggregations.

Prevent

• Treat RDD usage as technical debt in new code, and flag it in code review.