基于MapReduce与堆排序实现社交网络Top10被关注用户的方案问询

Great question! Let's break down how to tackle this Top 10 problem in MapReduce—since you already have the <userID, follower_count> pairs from your first job, you're halfway there. The key here is avoiding a full distributed sort (which is inefficient for large datasets) and instead using local Top-N aggregation + global Top-N finalization with priority queues. Here's a step-by-step breakdown:

Why Full Sort Isn't Ideal #

If you tried to sort all your <userID, follower_count> pairs globally, you'd force all data into a single reducer (since sorting requires a single ordered output), which would bottleneck performance on large datasets. Instead, we'll use a two-phase approach to keep computation distributed until the final step.

Phase 1: Local Top-10 in Mappers #

Each mapper processes a slice of your input data. We'll use a min-heap (priority queue) to track the top 10 users in its slice. A min-heap is perfect here because it lets us keep only the largest 10 values: the heap's top is the smallest of our current top 10, so we can quickly replace it if we encounter a larger value.

Mapper Code Example (Java) #

import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
import java.util.PriorityQueue;
import java.util.Comparator;

public class LocalTop10Mapper extends Mapper<LongWritable, Text, Text, Text> {
    private PriorityQueue<CountUserPair> minHeap;
    private static final int TOP_N = 10;

    // Initialize the min-heap with a comparator that sorts by follower count ascending
    @Override
    protected void setup(Context context) {
        minHeap = new PriorityQueue<>(TOP_N, Comparator.comparingInt(CountUserPair::getCount));
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // Parse input: assuming each line is "userID\tfollower_count"
        String[] parts = value.toString().split("\t");
        if (parts.length != 2) return; // Skip malformed lines

        String userId = parts[0];
        int followerCount = Integer.parseInt(parts[1]);

        // Update the min-heap
        if (minHeap.size() < TOP_N) {
            minHeap.add(new CountUserPair(followerCount, userId));
        } else {
            if (followerCount > minHeap.peek().getCount()) {
                minHeap.poll(); // Remove the smallest of the current top 10
                minHeap.add(new CountUserPair(followerCount, userId));
            }
        }
    }

    // Output the local top 10 when the mapper finishes processing its slice
    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        // Use a fixed key ("top10") to send all local top 10 results to a single reducer
        while (!minHeap.isEmpty()) {
            CountUserPair pair = minHeap.poll();
            context.write(new Text("top10"), new Text(pair.getUserId() + "\t" + pair.getCount()));
        }
    }

    // Helper class to store follower count and user ID (for heap operations)
    private static class CountUserPair {
        private final int count;
        private final String userId;

        public CountUserPair(int count, String userId) {
            this.count = count;
            this.userId = userId;
        }

        public int getCount() { return count; }
        public String getUserId() { return userId; }
    }
}

Phase 2: Global Top-10 in a Single Reducer #

All local top 10 results are sent to one reducer (thanks to the fixed "top10" key). The reducer uses another min-heap to aggregate these local results into the final global top 10.

Reducer Code Example (Java) #

import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
import java.util.PriorityQueue;
import java.util.ArrayList;
import java.util.Comparator;

public class GlobalTop10Reducer extends Reducer<Text, Text, Text, IntWritable> {
    private PriorityQueue<CountUserPair> minHeap;
    private static final int TOP_N = 10;

    @Override
    protected void setup(Context context) {
        minHeap = new PriorityQueue<>(TOP_N, Comparator.comparingInt(CountUserPair::getCount));
    }

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        for (Text value : values) {
            String[] parts = value.toString().split("\t");
            String userId = parts[0];
            int followerCount = Integer.parseInt(parts[1]);

            // Same heap logic as the mapper
            if (minHeap.size() < TOP_N) {
                minHeap.add(new CountUserPair(followerCount, userId));
            } else {
                if (followerCount > minHeap.peek().getCount()) {
                    minHeap.poll();
                    minHeap.add(new CountUserPair(followerCount, userId));
                }
            }
        }
    }

    // Output the final top 10 in descending order
    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        // Convert heap to a list and sort in reverse order (largest to smallest)
        ArrayList<CountUserPair> top10List = new ArrayList<>(minHeap);
        top10List.sort(Comparator.comparingInt(CountUserPair::getCount).reversed());

        for (CountUserPair pair : top10List) {
            context.write(new Text(pair.getUserId()), new IntWritable(pair.getCount()));
        }
    }

    // Reuse the same helper class as the mapper
    private static class CountUserPair {
        private final int count;
        private final String userId;

        public CountUserPair(int count, String userId) {
            this.count = count;
            this.userId = userId;
        }

        public int getCount() { return count; }
        public String getUserId() { return userId; }
    }
}

Key Notes #

1. Why Min-Heap? A min-heap ensures we only keep the largest N values with minimal memory usage (O(N) space). If we used a max-heap, we'd have to store all values and then extract the top 10, which is less efficient.
2. Single Reducer? Since we're only aggregating number_of_mappers * 10 entries, a single reducer won't be a bottleneck—this is way better than pushing the entire dataset through one reducer for a full sort.
3. Error Handling: Add checks for malformed input lines (like non-numeric follower counts) to avoid crashes.
4. Custom Writables: If you need to optimize serialization, you can create a custom Writable class for CountUserPair instead of using Text strings, but for small N, the overhead is negligible.

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