 (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)))
+;; Returns a stopword filtered seq of
+;; [doc-id term term-frequency doc-terms-count] tuples
+(f/defsparkfn gen-docid-term-tuples [doc-tuple]
+  (let [[doc-id content] doc-tuple
+        terms (filter #(not (contains? stopwords %))
+                      (clojure.string/split content #" "))
+        doc-terms-count (count terms)
+        term-frequencies (frequencies terms)]
+    (map (fn [term] [doc-id term (term-frequencies term) doc-terms-count])
+         (distinct 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 calc-idf [doc-count]
+  (f/fn [[term tuple-seq]]
+    (let [df (count tuple-seq)]
+      [term (Math/log (/ doc-count (+ 1.0 df)))])))
 
 (defn -main [& args]
   (try


          doc-data (f/parallelize sc documents)

          ;; stopword filtered RDD of (document_id, term) 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)
          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/cache)

          ;; total number of documents in corpus


          ;; 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)))))
