 
 (defn trivial-join-mapper
   [tag input]
-  (->> input mr/vals
-       (r/map (fn [line]
-                (let [[key val] (str/split line #"\s")]
-                  [[(Long/parseLong key) tag] val])))))
+  (r/map (fn [line]
+           (let [[key val] (str/split line #"\s")]
+             [[(Long/parseLong key) tag] val]))
+         input))
 
 (defn trivial-join-partitioner
   ^long [[key] _ ^long nparts]
   (-> key hash (mod nparts)))
 
 (defn trivial-join-reducer
-  [input]
-  (->> input mr/keyvalgroups
-       (r/mapcat (fn [[[id] vals]]
-                   (let [vals (into [] vals)
-                         left (first vals)]
-                     (r/map #(-> [id left %]) (rest vals)))))
-       (mr/sink-as :keys)))
+  {::mr/source-as :keyvalgroups}
+  [input]
+  (r/mapcat (fn [[[id] vals]]
+              (let [vals (into [] vals)
+                    left (first vals)]
+                (r/map #(-> [id left %]) (rest vals))))
+            input))
 
 (defn trivial-join
   [conf left right dsink]

(defn word-count-mapper
  [input]
  (let [wc (.getCounter mr/*context* "word-count" "words")]
    (->> input mr/vals
         (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 (->> [["apple banana banana"]
                   ["carrot apple"]
                   ["apple"]]
                  (mem/dseq :keys))
        dsink (seqf/dsink [Text LongWritable] outpath)
        [result] (word-count (th/config) dseq dsink)]
    (is (= {"apple" 3, "banana" 2, "carrot" 1}
