title: Spark - Reduce operation taking too long
body: <p>I'm making an application with Spark that will run some topic extration algorithms. For that, first I need to make some preprocessing, extracting the document-term matrix by the end. Ive could done that, but for a (not that much) big collection of documents (only 2 thousand, 5MB), this proccess is taking forever.</p>
  
  <p>So, debugging, Ive found where the program kinda stucks, and it's in a reduce operation. What I'm doing in this part of the code is counting how many times each term occurs on the collection, so first I done a "map", couting it for each rdd, and them I "reduce" it, saving the result inside a hashmap. The map operation is very fast, but in the reduce, its splitting the operation in 40 blocks, and each block takes 5~10 minutes to proccess.</p>
  
  <p>So I'm trying to figure out what I'm doing wrong, or if reduce operations are that much costly.</p>
  
  <p>SparkConf: Standalone mode, using local[2]. I've tried to use it as "spark://master:7077", and it worked, but still the same slowness.</p>
  
  <p>Code:</p>
  
  <p><em>"filesIn" is a JavaPairRDD where the key is the file path and the value is the content of the file.
  So, first the map, where I take this "filesIn", split the words, and count their frequency (in that case doesn't matter what document is)
  And then the reduce, where I create a HashMap (term, freq).</em></p>
  
  <pre><code>JavaRDD&lt;HashMap&lt;String, Integer&gt;&gt; termDF_ = filesIn.map(new Function&lt;Tuple2&lt;String, String&gt;, HashMap&lt;String, Integer&gt;&gt;() {
  
          @Override
          public HashMap&lt;String, Integer&gt; call(Tuple2&lt;String, String&gt; t) throws Exception {
              String[] allWords = t._2.split(" ");
  
              HashMap&lt;String, Double&gt; hashTermFreq = new HashMap&lt;String, Double&gt;();
              ArrayList&lt;String&gt; words = new ArrayList&lt;String&gt;();
              ArrayList&lt;String&gt; terms = new ArrayList&lt;String&gt;();
              HashMap&lt;String, Integer&gt; termDF = new HashMap&lt;String, Integer&gt;();
  
              for (String term : allWords) {
  
                  if (hashTermFreq.containsKey(term)) {
                      Double freq = hashTermFreq.get(term);
                      hashTermFreq.put(term, freq + 1);
                  } else {
                      if (term.length() &gt; 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&lt;String, Integer&gt; termDF = termDF_.reduce(new Function2&lt;HashMap&lt;String, Integer&gt;, HashMap&lt;String, Integer&gt;, HashMap&lt;String, Integer&gt;&gt;() {
  
          @Override
          public HashMap&lt;String, Integer&gt; call(HashMap&lt;String, Integer&gt; t1, HashMap&lt;String, Integer&gt; t2) throws Exception {
              HashMap&lt;String, Integer&gt; result = new HashMap&lt;String, Integer&gt;();
  
              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;
          }
      });
  </code></pre>
  
  <p>Thanks!</p>