PySpark自定义累加器报'int is not iterable'错误，寻求技术协助

Hey there, let's break down why you're hitting this error and fix your code step by step.

What's Causing the Error? #

Looking at your code, the main issue lies in two key places:

1. The addInPlace method uses v1.union(v2), but v2 is a single integer (from your RDD of xrange(10)). The union() method expects an iterable (like another set), not a single int—this is exactly why you get the "int is not iterable" error.
2. Your zero method creates a set containing the initial value, but for an accumulator meant to track all seen IDs, we need the starting state to be an empty set, not a set with an initial element.

Corrected Code #

Here's the fixed version that correctly collects all unique integers from your RDD into a set accumulator:

import findspark
findspark.init()
from pyspark import AccumulatorParam, SparkContext

sc = SparkContext('local','local')
rdd = sc.parallelize(range(10))  # Switched to range for Python 3 compatibility

class SetAccumulatorParam(AccumulatorParam):
    def zero(self, initial_value):
        # Return an empty set as the starting zero state
        return set() if not isinstance(initial_value, set) else initial_value.copy()
    
    def addInPlace(self, current_set, new_element):
        # Add the single integer element to the existing set (no union needed)
        current_set.add(new_element)
        return current_set

# Initialize the accumulator with an empty set
ids_seen = sc.accumulator(set(), SetAccumulatorParam())

# Function to update the accumulator with each RDD element
def update_accumulator(num):
    global ids_seen
    ids_seen.add(num)

# Apply the update to every element in the RDD
rdd.foreach(update_accumulator)

# Check the final result
print("IDs seen:", ids_seen.value)
sc.stop()

Key Fixes Explained #

• addInPlace Method: Swapped union() for add() because we're inserting individual integers into the set, not merging two sets. add() is the correct method for single-element insertion.
• zero Method: Now returns an empty set as the initial state, which makes logical sense for tracking "seen" IDs (we start with no IDs recorded).
• Accumulator Initialization: We initialize the accumulator with an empty set, matching the type our custom parameter class expects.

Testing the Code #

When you run this, you'll get the output:

IDs seen: {0, 1, 2, 3, 4, 5, 6, 7, 8, 9}

This correctly collects all integers from your RDD into the accumulator set without any errors.

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