 
           doc-data (f/parallelize sc documents)
 
-          ;; stopword filtered RDD of (document_id, term) tuples
+          ;; stopword filtered RDD of [doc-id term term-freq doc-terms-count] tuples
           doc-term-seq (-> doc-data
                            (f/flat-map gen-docid-term-tuples)
-                           (f/filter filter-stopwords)
                            f/cache)
 
-          ;; tf per document, that is, tf(term, document)
+          ;; RDD of term-frequency tuples: [term [doc-id tf]]
+          ;; where tf is per document, that is, tf(term, document)
           tf-by-doc (-> doc-term-seq
-                        (f/group-by (f/fn [[doc-id term]] [doc-id term]))
-                        ;; (raw) number of times a term appears in a document
-                        (f/map (f/fn [[[doc-id term] val-seq]] [doc-id [term (count val-seq)]]))
-                        f/group-by-key
-                        (f/flat-map term-freq-per-doc)
+                        (f/map (f/fn [[doc-id term term-freq doc-terms-count]]
+                                     [term [doc-id (double (/ term-freq doc-terms-count))]]))
                         f/cache)
 
           ;; total number of documents in corpus

(def stopwords #{"a" "all" "and" "any" "are" "is" "in" "of" "on"
                 "or" "our" "so" "this" "the" "that" "to" "we"})

;; Returns a seq of (doc_id, term) tuples
(f/defsparkfn gen-docid-term-tuples [x]
  (let [[doc-id content] x
        terms (clojure.string/split content #" ")]
    (map (fn [term] [doc-id term]) terms)))

;; Filters (doc_id, term) tuples if term is a stopword
(f/defsparkfn filter-stopwords
  [[_ term]]
  (not (contains? stopwords term)))

;; Returns a seq of term-frequency tuples: (term, (doc_id, tf))
(f/defsparkfn term-freq-per-doc
  [[doc-id term-freq-seq]]
  (let [terms-count (reduce (fn [accum [_ term-freq]]
                              (+ accum term-freq))
                            0
                            term-freq-seq)]
    (map (fn [[term term-freq]]
           [term [doc-id (double (/ term-freq terms-count))]])
         term-freq-seq)))

(defn -main [& args]
  (try


          ;; idf of terms, that is, idf(term)
          idf-by-term (-> doc-term-seq
                          (f/group-by (f/fn [[_ term _ _]] term))
                          (f/map (calc-idf num-docs))
                          f/cache)

          ;; tf-idf of terms, that is, tf(term, document) x idf(term)
          tfidf-by-term (-> (f/join tf-by-doc idf-by-term)
                            (f/map (f/fn [[term [[doc-id tf] idf]]]
                                         [doc-id term (* tf idf)]))
                            f/cache)
          ]
      (->> tfidf-by-term
           f/collect
           ((partial sort-by last >))
           (take 10)
           clojure.pprint/pprint))
    (catch Exception e
      (println (.printStackTrace e)))))
