火花 - 减少操作时间过长

Question

我正在使用Spark制作一个应用程序，该应用程序将运行一些主题额外算法。为此，首先我需要进行一些预处理，最后提取文档术语矩阵。我可以做到这一点，但对于(不是那么多(大文档集合(只有 2000,5MB(，这个过程需要很长时间。

所以，在调试中，我找到了程序有点卡住的地方，它处于reduce操作中。我在代码的这一部分所做的是计算每个术语在集合中出现的次数，所以首先我做了一个"map"，为每个rdd搜索它，然后他们"减少"它，将结果保存在哈希map中。map操作非常快，但在reduce中，它将操作拆分为40个块，每个块需要5~10分钟才能处理。

所以我试图弄清楚我做错了什么，或者减少操作是否成本很高。

SparkConf：独立模式，使用本地[2]。我尝试将其用作"spark://master:7077"，它奏效了，但仍然一样慢。

法典：

"filesIn"是一个JavaPairRDD，其中键是文件路径，值是文件的内容。所以，首先是地图，我把这个"fileIn"放在其中，拆分单词，并计算它们的频率(在这种情况下，是什么文档并不重要(然后是reduce，我在其中创建了一个HashMap(术语，频率(。

JavaRDD<HashMap<String, Integer>> termDF_ = filesIn.map(new Function<Tuple2<String, String>, HashMap<String, Integer>>() {
        @Override
        public HashMap<String, Integer> call(Tuple2<String, String> t) throws Exception {
            String[] allWords = t._2.split(" ");
            HashMap<String, Double> hashTermFreq = new HashMap<String, Double>();
            ArrayList<String> words = new ArrayList<String>();
            ArrayList<String> terms = new ArrayList<String>();
            HashMap<String, Integer> termDF = new HashMap<String, Integer>();
            for (String term : allWords) {
                if (hashTermFreq.containsKey(term)) {
                    Double freq = hashTermFreq.get(term);
                    hashTermFreq.put(term, freq + 1);
                } else {
                    if (term.length() > 1) {
                        hashTermFreq.put(term, 1.0);
                        if (!terms.contains(term)) {
                            terms.add(term);
                        }
                        if (!words.contains(term)) {
                            words.add(term);
                            if (termDF.containsKey(term)) {
                                int value = termDF.get(term);
                                value++;
                                termDF.put(term, value);
                            } else {
                                termDF.put(term, 1);
                            }
                        }
                    }
                }
            }
            return termDF;
        }
    });
 HashMap<String, Integer> termDF = termDF_.reduce(new Function2<HashMap<String, Integer>, HashMap<String, Integer>, HashMap<String, Integer>>() {
        @Override
        public HashMap<String, Integer> call(HashMap<String, Integer> t1, HashMap<String, Integer> t2) throws Exception {
            HashMap<String, Integer> result = new HashMap<String, Integer>();
            Iterator iterator = t1.keySet().iterator();
            while (iterator.hasNext()) {
                String key = (String) iterator.next();
                if (result.containsKey(key) == false) {
                    result.put(key, t1.get(key));
                } else {
                    result.put(key, result.get(key) + 1);
                }
            }
            iterator = t2.keySet().iterator();
            while (iterator.hasNext()) {
                String key = (String) iterator.next();
                if (result.containsKey(key) == false) {
                    result.put(key, t2.get(key));
                } else {
                    result.put(key, result.get(key) + 1);
                }
            }
            return result;
        }
    });

谢谢！

Answer 1

好的，所以就在我的头顶上：

• Spark 转换是惰性的。这意味着在调用后续reduce操作之前不会执行map，因此您描述的慢reduce很可能是慢map + reduce
• ArrayList.contains是O(N(，所以所有这些words.contains和terms.contains效率极低
• map逻辑闻起来很腥。特别：
  • 如果已经看到术语，您将永远不会进入else分支
  • 乍一看，words和terms应该具有完全相同的内容，并且应该等同于hashTermFreq键或termDF键。
  • 看起来 termDF 中的值只能取值 1。如果这是您想要的并且您忽略了频率，那么创建hashTermFreq的意义何在？
• 这里实现reduce阶段意味着低效的线性扫描，对象在数据上不断增长，而您真正想要的是reduceByKey。

使用 Scala 作为伪代码，您的整个代码可以有效地表示如下：

val termDF = filesIn.flatMap{
  case (_, text) => 
    text.split(" ") // Split
    .toSet // Take unique terms 
    .filter(_.size > 1) // Remove single characters
    .map(term => (term, 1))} // map to pairs
  .reduceByKey(_ + _) // Reduce by key
termDF.collectAsMap // Optionally

最后，看起来你正在重新发明轮子。至少您需要的一些工具已经在mllib.feature或ml.feature中实现