How do I fix "Task not serializable" in Spark?

Summary

• "Task not serializable" is a SparkException thrown when Spark can't serialize the function you pass to a transformation together with the variables and methods it captures (the closure). The exception's cause names the offending class, typically java.io.NotSerializableException.
• The usual fix is to keep the unserializable object out of the closure. Copy the field you need into a local variable instead of referencing this, or create connections and clients on the executor inside foreachPartition or mapPartitions.
• On the JVM, a class that has to travel with the closure must implement java.io.Serializable. PySpark serializes functions with cloudpickle, and a failure typically surfaces as pickle.PicklingError with the message "Could not serialize object".
• Changing a driver variable inside a closure is a different bug: there is no exception, but the driver's value doesn't change. Use an accumulator for that, and a broadcast variable for read-only lookup data.

How do I fix "Task not serializable" in Spark?

Spark computes a closure for every task: "those variables and methods which must be visible for the executor to perform its computations". That "closure is serialized and sent to each executor". When you pass a function to an RDD transformation such as map, Spark serializes the cleaned closure on the spot as a check, and throws SparkException: Task not serializable if that fails. So the fix is to change what the closure captures, not to retry the job.

How do I find the object that can't be serialized?

Read the cause under the SparkException. On the JVM it is usually java.io.NotSerializableException plus a class name, which the Java docs describe as "Thrown when an instance is required to have a Serializable interface". The Spark tuning guide says Java serialization "can work with any class you create that implements java.io.Serializable". PySpark pickles functions with cloudpickle, and a failure typically surfaces as pickle.PicklingError, with a message that starts "Could not serialize object".

Which cause matches your code?

Pattern	Why it breaks	Fix
The closure reads a field or calls a method of an enclosing instance	"the whole object needs to be sent to the cluster"	Copy the field into a local variable, or make the class java.io.Serializable
The closure captures a connection or client created on the driver	"Such connection objects are rarely transferable across machines"	Create it on the executor, once per partition, with foreachPartition or mapPartitions
A PySpark function uses SparkContext	"SparkContext can only be used on the driver, not in code that runs on workers"	Do that work on the driver
The closure updates a driver variable such as a counter	Executors update copies, so the driver's value "will still be zero"	Use an accumulator
Many tasks ship the same large read-only data	Broadcast variables let you "keep a read-only variable cached on each machine rather than shipping a copy of it with tasks"	Use a broadcast variable

What do the fixes look like?

The Spark RDD guide's Scala example copies the field so the closure no longer references this:

class MyClass {
  val field = "Hello"
  def doStuff(rdd: RDD[String]): RDD[String] = {
    // rdd.map(x => field + x) would reference `this`, so all of MyClass is serialized
    val field_ = this.field
    rdd.map(x => field_ + x)
  }
}

The Spark Streaming guide's pattern for external systems creates the connection on the worker, once per partition, because per-record connections add overhead:

def sendPartition(iter):
    connection = createNewConnection()  # runs on the executor
    for record in iter:
        connection.send(record)
    connection.close()

rdd.foreachPartition(sendPartition)

For counters, the guide's accumulator example updates the accumulator inside foreach. Spark guarantees that updates made inside actions are applied once, while updates inside transformations may be applied more than once if tasks or stages are re-executed.

accum = sc.accumulator(0)
rdd.foreach(lambda x: accum.add(x))
print(accum.value)  # only the driver can read the value

How does Databricks handle this?

In Databricks Connect for Python, UDFs "are serialized by Databricks Connect and sent to the server as part of the request", so "the Python version of the client must match the Python version on the Databricks compute". For Scala UDFs, "the compiled classes and JARs must be uploaded to the cluster using the addCompiledArtifacts() API". On serverless compute, "Spark RDD APIs are not supported", so the RDD examples on this page don't apply there.

FAQs

Why does the error appear before any job runs?

RDD.map calls sc.clean(f) as soon as you define the transformation, and the closure cleaner checks serializability by default. The failure surfaces at that line, not inside a running task.

How do I test whether a Python function will serialize?

PySpark's serializer imports cloudpickle (from pyspark import cloudpickle), so test with cloudpickle.dumps(obj). Plain pickle cannot pickle lambdas, but cloudpickle "supports pickling for lambda functions". It "can only be used to send objects between the exact same version of Python", so keep the Python version the same on the driver and the workers.

Why did my counter work in local mode but not on a cluster?

In local mode the foreach function "will actually execute within the same JVM as the driver and will reference the same original counter". The guide calls that accidental: such code "will not behave as expected in distributed mode".