基于MapReduce的KMeans算法PySpark实现遇阻求助

Let's break down the issues in your code and fix them step by step—since the ValueError: not enough values to unpack (expected 2, got 1) is just the tip of the iceberg when implementing MapReduce-style KMeans correctly in PySpark.

Key Issues in Your Current Code #

1. Mapper Isn't Generating Valid (Key, Value) Pairs #

Your mapper function uses yield to emit (closest_centroid, datapoint) pairs, but you're using map() instead of flatMap() on the RDD. map() wraps each generator from mapper into a single RDD element, while flatMap() flattens all yielded pairs into individual (key, value) entries. This is why reduceByKey can't unpack values—your RDD elements weren't valid tuples, they were generator objects.

2. Misused reduceByKeyLocally & Broken Combiner/Reducer Logic #

reduceByKeyLocally expects a binary function (takes two values of the same type, returns one value of that type) to aggregate values for the same key. Your current combiner/Reducer functions are structured to process iterators, which doesn't match what reduceByKey passes to them. Additionally:

• You used list concatenation (sum = sum + vec) instead of element-wise addition for vector sums (this would append lists instead of adding their values).
• The functions used yield instead of returning a single aggregated result per key.
• Wrapping vectors in Row added unnecessary complexity; lists/tuples are simpler for vector operations.

3. Flawed Centroid Update & Convergence Check #

Your iteration loop had incorrect comparisons (comparing reduced[i][1] to reduced[i][0] doesn't make sense) and mishandled how new centroids are assigned to the centers list.

Corrected Implementation #

Here's the fixed code with explanations for each change:

from pyspark.sql import Row

def find_dist(x, y):
    """Calculate squared Euclidean distance between two vectors (avoids sqrt for efficiency)"""
    return sum((xi - yi)**2 for xi, yi in zip(x, y))

def mapper(centers, datapoint):
    """Map each datapoint to (closest_centroid_tuple, (datapoint_vector, 1))"""
    # Convert Row to list for easier arithmetic operations
    dp_vec = list(datapoint)
    # Initialize with first centroid
    min_dist = find_dist(dp_vec, centers[0])
    closest_cent = tuple(centers[0])  # Tuples are hashable, so they work as RDD keys
    
    # Find the closest centroid
    for cent in centers[1:]:
        curr_dist = find_dist(dp_vec, cent)
        if curr_dist < min_dist:
            min_dist = curr_dist
            closest_cent = tuple(cent)
    
    # Return valid (key, value) tuple for aggregation
    return (closest_cent, (dp_vec, 1))

def combiner(val1, val2):
    """Combine two (vector_sum, count) pairs into one aggregated pair"""
    # val1 = (sum_vec1, count1), val2 = (sum_vec2, count2)
    sum_vec = [v1 + v2 for v1, v2 in zip(val1[0], val2[0])]
    total_count = val1[1] + val2[1]
    return (sum_vec, total_count)

def reducer(val1, val2):
    """Reduce multiple combined pairs into a final (sum_vec, total_count) pair"""
    # Logic matches combiner since combining and reducing are identical here
    sum_vec = [v1 + v2 for v1, v2 in zip(val1[0], val2[0])]
    total_count = val1[1] + val2[1]
    return (sum_vec, total_count)

def kmeans_fit(data, k, max_iter):
    # Initialize centroids as lists (easier for arithmetic operations)
    centers = [list(row) for row in data.rdd.takeSample(False, k, seed=42)]
    
    for iter_num in range(max_iter):
        # Step 1: Map each datapoint to its closest centroid, with vector and count
        mapped_rdd = data.rdd.map(lambda dp: mapper(centers, dp))
        
        # Step 2: Combine and reduce to get sum of vectors and count per centroid
        aggregated = mapped_rdd.reduceByKey(reducer).collect()
        
        # Step 3: Calculate new centroids and check for convergence
        new_centers = []
        converged = True
        epsilon = 1e-6  # Handle floating-point precision issues
        
        for cent_key, (sum_vec, count) in aggregated:
            # Compute average to get new centroid
            new_cent = [s / count for s in sum_vec]
            new_centers.append(new_cent)
            
            # Find the matching old centroid and check if it changed
            old_cent = next(c for c in centers if tuple(c) == cent_key)
            if not all(abs(n - o) < epsilon for n, o in zip(new_cent, old_cent)):
                converged = False
        
        # Update centers for next iteration
        centers = new_centers
        
        # Exit early if centroids stop changing
        if converged:
            print(f"Converged after {iter_num + 1} iterations")
            break
    
    return centers

# Load data and run the algorithm
data = spark.read.parquet("/mnt/ddscoursedatabricksstg/ddscoursedatabricksdata/random_data.parquet")
final_centers = kmeans_fit(data, 5, 10)
print("Final Centroids:", final_centers)

Key Improvements Explained #

1. Mapper Fix:

   • Used map() (instead of flatMap() here, since each datapoint emits exactly one pair) to ensure valid (key, value) tuples in the RDD.
   • Converted centroids to tuples to use as hashable RDD keys (lists can't be keys in PySpark).
   • Emitted values as (datapoint_vector, 1) to simplify tracking sum and count during aggregation.

2. Combiner/Reducer Fix:

   • Both functions now take two aggregated values and return a single combined result, matching reduceByKey's requirements.
   • Used element-wise addition for vector sums instead of list concatenation.
   • Removed unnecessary Row wrappers to streamline vector operations.

3. Convergence Check:

   • Added a small epsilon (1e-6) to handle floating-point precision issues when comparing old and new centroids.
   • Properly maps old centroids to new ones using their tuple keys, ensuring accurate convergence checks.
   • Added early exit logic to stop iterating once centroids stabilize, optimizing performance.

4. Simplified Distance Calculation:

   • Rewrote find_dist using a generator expression for cleaner, more efficient code (and skipped the square root since we only need relative distances).

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