java - 如何避免使用 ConcurrentHashMap

Question

我在 Hadoop 的 Reducer 类的 run() 方法中编写了这段代码

@Override
    public void run(Context context) throws IOException, InterruptedException {
        setup(context);

        ConcurrentHashMap<String, HashSet<Text>> map = new ConcurrentHashMap<String, HashSet<Text>>();

        while (context.nextKey()) {
            String line = context.getCurrentKey().toString();
            HashSet<Text> values = new HashSet<Text>();
            for (Text t : context.getValues()) {
                values.add(new Text(t));
            }

            map.put(line, new HashSet<Text>());
            for (Text t : values) {
                map.get(line).add(new Text(t));
            }
        }

        ConcurrentHashMap<String, HashSet<Text>> newMap = new ConcurrentHashMap<String, HashSet<Text>>();

        for (String keyToMerge : map.keySet()) {
            String[] keyToMergeTokens = keyToMerge.split(",");
            for (String key : map.keySet()) {
                String[] keyTokens = key.split(",");
                if (keyToMergeTokens[keyToMergeTokens.length - 1].equals(keyTokens[0])) {
                    String newKey = keyToMerge;
                    for (int i = 1; i < keyTokens.length; i++) {
                        newKey += "," + keyTokens[i];
                    }
                    if (!newMap.contains(newKey)) {
                        newMap.put(newKey, new HashSet<Text>());
                        for (Text t : map.get(keyToMerge)) {
                            newMap.get(newKey).add(new Text(t));
                        }
                    }
                    for (Text t : map.get(key)) {
                        newMap.get(newKey).add(new Text(t));
                    }
                }
            }


        //call the reducers
        for (String key : newMap.keySet()) {
            reduce(new Text(key), newMap.get(key), context);
        }

        cleanup(context);
    }

我的问题是，即使我的输入太小，由于 newMap.put() 调用，也需要 30 分钟才能运行。如果我将此命令放在注释中，那么它会快速运行而不会出现任何问题。如您所见，我使用了 ConcurrentHashMap。我不想使用它，因为我认为 run() 在每台机器上只被调用一次(它不会同时运行)所以我不会对简单的 HashMap 有任何问题但是如果我用简单的替换 concurrentHashMap HashMap 我收到一个错误 (concurrentModificationError)。有没有人知道如何让它在没有任何延迟的情况下工作？提前致谢!

*java6*hadoop 1.2.1

Answer 1

我不知道它是否能解决您的性能问题，但我看到您正在做一件效率低下的事情:

newMap.put(newKey, new HashSet<Text>());
for (Text t : map.get(keyToMerge)) {
    newMap.get(newKey).add(new Text(t));
}

将 HashSet 保存在变量中而不是在 newMap 中搜索它会更有效:

HashSet<Text> newSet = new HashSet<Text>();
newMap.put(newKey, newSet);
for (Text t : map.get(keyToMerge)) {
    newSet.add(new Text(t));
}

您正在做的另一件低效的事情是创建一个 HashSet 值，然后创建另一个相同的 HashSet 以放入映射中。由于原始的 HashSet (values) 不再被使用，因此您毫无理由地构造所有这些 Text 对象。

代替:

    while (context.nextKey()) {
        String line = context.getCurrentKey().toString();
        HashSet<Text> values = new HashSet<Text>();
        for (Text t : context.getValues()) {
            values.add(new Text(t));
        }

        map.put(line, new HashSet<Text>());
        for (Text t : values) {
            map.get(line).add(new Text(t));
        }
    }

你可以简单地写:

    while (context.nextKey()) {
        String line = context.getCurrentKey().toString();
        HashSet<Text> values = new HashSet<Text>();
        for (Text t : context.getValues()) {
            values.add(new Text(t));
        }

        map.put(line, values);
    }

编辑:

我刚刚看到您作为答案发布的附加代码(来自您的 cleanup() 方法):

    //clear map
    for (String s : map.keySet()) {
        map.remove(s);
    }
    map = null;

    //clear newMap
    for (String s : newMap.keySet()) {
        newMap.remove(s);
    }
    newMap = null;

此代码为您提供 ConcurrentModificationError 的原因是 foreach 循环不支持修改您正在迭代的集合。

为了克服这个问题，您可以使用迭代器:

    //clear map
    Iterator<Map.Entry<String, HashSet<Text>>> iter1 = map.entrySet ().iterator ();
    while (iter1.hasNext()) {
        Map.Entry<String, HashSet<Text>> entry = iter1.next();
        iter1.remove();
    }
    map = null;

    //clear newMap
    Iterator<Map.Entry<String, HashSet<Text>>> iter2 = newMap.entrySet ().iterator ();
    while (iter2.hasNext()) {
        Map.Entry<String, HashSet<Text>> entry = iter2.next();
        iter2.remove();
    }
    newMap = null;

也就是说，您实际上不必分别删除每个项目。你可以简单地写

map = null;
newMap = null;

当您删除对 map 的引用时，垃圾收集器可以对它们进行垃圾收集。从 map 中删除项目没有任何区别。