549. 最常使用的k个单词(Map Reduce)

描述

使用 map reduce 框架查找最常使用的 k 个单词.
mapper 的 key 为文档的 id, 值是文档的内容, 文档中的单词由空格分割.
对于 reducer，应该输出最多为 k 个 key-value 对, 包括最常用的 k 个单词以及他们在当前 reducer 中的使用频率.评判系统会合并不同的 reducer 中的结果以得到 全局 最常使用的 k 个单词, 所以你不需要关注这一环节. k 在 TopK 类的构造器中给出.

注意事项

如果单词有相同的使用频率,那么按照字母排序

样例

给出文档 A =
lintcode is the best online judge
I love lintcode
以及文档 B =
lintcode is an online judge for coding interview
you can test your code online at lintcode
最常用的2个单词以及他们的使用频率应该为:
lintcode, 4
online, 3

代码

/**
 * Definition of OutputCollector:
 * class OutputCollector<K, V> {
 *     public void collect(K key, V value);
 *         // Adds a key/value pair to the output buffer
 * }
 * Definition of Document:
 * class Document {
 *     public int id;
 *     public String content;
 * }
 */
class Pair {
    String key;
    int value;
    
    Pair(String key, int value) {
        this.key = key;
        this.value = value;
    }
}

public class TopKFrequentWords {

    public static class Map {
        public void map(String _, Document value,
                        OutputCollector<String, Integer> output) {
            // Write your code here
            // Output the results into output buffer.
            // Ps. output.collect(String key, int value);
            int id = value.id;
            String content = value.content;
            String[] words = content.split(" ");
            for (String word : words) {
                if (word.length() > 0) {
                    output.collect(word, 1);
                }
            }
        }
    }

    public static class Reduce {
        private PriorityQueue<Pair> Q = null;
        private int k;

        // 从小到大排序
        private Comparator<Pair> pairComparator = new Comparator<Pair>() {
            public int compare(Pair left, Pair right) {
                if (left.value != right.value) {
                    return left.value - right.value;
                }
                return right.key.compareTo(left.key);
            }
        };

        public void setup(int k) {
            // initialize your data structure here
            this.k = k;
            Q = new PriorityQueue<Pair>(k, pairComparator);
        }   

        public void reduce(String key, Iterator<Integer> values) {
            // Write your code here
            int sum = 0;
            while (values.hasNext()) {
                    sum += values.next();
            }

            Pair pair = new Pair(key, sum);
            if (Q.size() < k) {
                Q.add(pair);
            } else {
                Pair peak = Q.peek();
                // 新加入的数比最小堆堆顶元素大，替换
                if (pairComparator.compare(pair, peak) > 0) {
                    Q.poll();
                    Q.add(pair);
                }
            }
        }

        public void cleanup(OutputCollector<String, Integer> output) {
            // Output the top k pairs <word, times> into output buffer.
            // Ps. output.collect(String key, Integer value);
            List<Pair> pairs = new ArrayList<Pair>();
            while (!Q.isEmpty()) {
                pairs.add(Q.poll());
            }

            // 先抛出来的是出现频率最低的，题目要求根据频率由高到低输出
            // reverse result
            int n = pairs.size();
            for (int i = n - 1; i >= 0; --i) {
                Pair pair = pairs.get(i);
                // 已经定义调用 collect 方法将键值对输出
                output.collect(pair.key, pair.value);
            }
        }
    }
}