Spark + DEAP示例运行失败：集群并行化DEAP遇属性获取错误

I’ve run into this exact issue when parallelizing DEAP workflows on Spark, so let’s walk through why it happens and how to fix it.

Why This Error Occurs #

Spark runs tasks across multiple executor nodes, each with its own isolated Python environment. When you define creator.Individual (and other custom DEAP classes) on the driver node, those are dynamically generated classes that don’t exist in the executor environments by default. When Spark tries to send your DEAP individuals to executors or run code that references Individual, the executor can’t find the class definition—hence the "Can't get attribute" error.

Practical Fixes #

1. Distribute DEAP Class Definitions to All Executors

The simplest fix is to package your DEAP setup code (where you create Fitness and Individual classes) into a separate Python file, then tell Spark to distribute this file to every executor.

• First, create a file like deap_config.py with your class creation logic:

  from deap import base, creator, tools
  
  # Copy your exact creator setup here
  creator.create("FitnessMax", base.Fitness, weights=(1.0,))
  creator.create("Individual", list, fitness=creator.FitnessMax)
  
  # Add any other toolbox setup you need (e.g., toolbox.register(...) calls)
  

• In your main Spark driver code, add this file to Spark’s distributed files before running any parallel tasks:

  from pyspark import SparkContext
  
  sc = SparkContext(appName="DEAP-Spark")
  sc.addPyFile("deap_config.py")  # Ensures every executor gets this file
  

• In your parallel functions (like map or flatMap), import the creator from this distributed file:

  def evaluate_individual(ind_data):
      # Import the creator from the distributed config file
      from deap_config import creator
      # Reconstruct the Individual instance if needed
      ind = creator.Individual(ind_data)
      # Your evaluation logic here
      fitness_value = ...  # Calculate fitness
      return (ind_data, fitness_value)
  

2. Avoid Serializing Dynamic DEAP Instances Directly

Instead of sending full Individual objects from driver to executors, send the raw underlying data (like the list that makes up the individual). Then, reconstruct the Individual instance on the executor side using the distributed deap_config.py setup. This avoids trying to serialize the dynamically generated class, which Pickle (Spark’s default serializer) struggles with.

3. Use CloudPickle for Serialization

Spark’s default Python serializer can’t handle some dynamic Python objects. Switching to CloudPickle (which supports more complex serialization) can resolve this without changing your class setup.

Update your Spark configuration to use CloudPickle:

from pyspark import SparkConf, SparkContext

conf = SparkConf()
conf.set("spark.python.serializer", "pyspark.serializers.CloudPickleSerializer")
sc = SparkContext(conf=conf, appName="DEAP-Spark")

Final Notes #

Make sure that any DEAP-related imports and class definitions are either:

• Included in the distributed deap_config.py file, or
• Re-executed inside the parallel task functions (so executors create the classes themselves)

This ensures every executor has access to the exact Individual class definition your code references.

内容的提问来源于stack exchange，提问作者Ryan