 (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


(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])))))

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

(defn trivial-join
  [conf left right dsink]

         (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]
