 (defn word-count-mapper
   [input]
   (let [wc (.getCounter mr/*context* "word-count" "words")]
-    (->> input mr/vals
+    (->> input
          (r/mapcat #(str/split % #"\s+"))
          (r/map (fn [word]
                   (returning word


(deftest test-word-count-mem
  (let [outpath (doto (fs/path "tmp/word-count-output") fs/path-delete)
        dseq (mem/dseq [[nil "apple banana banana"]
                        [nil "carrot apple"]
                        [nil "apple"]])
        dsink (seqf/dsink [Text LongWritable] outpath)
        [result] (word-count (th/config) dseq dsink)]
    (is (= {"apple" 3, "banana" 2, "carrot" 1}

         (r/map #(-> [% 1])))))

(defn word-count-reducer
  {::mr/source-as :keyvalgroups}
  [input]
  (r/map (fn [[word counts]]
           [word (r/reduce + 0 counts)])
         input))

(defn word-count
  [conf dseq dsink]
