     (.setInputFormat job SequenceFileInputFormat)
     (.setOutputFormat job SequenceFileOutputFormat)
     (SequenceFileOutputFormat/setOutputCompressionType job SequenceFile$CompressionType/BLOCK)
-    (.setOutputKeyClass job IntWritable)
+    (.setOutputKeyClass job Text)
     (.setOutputValueClass job Text)
     job))
 


(import-hadoop)

(defn default-jobconf [this]
  (let [conf (.getConf this)
        job (JobConf. conf (.getClass this))]

  [stage corpus]
  (str (job-path stage corpus) "/docids.tsv.gz"))

(defn standard-map [map-fn this wkey wvalue output reporter]
  (let [key (read-string (str wkey))
        value (read-string (str wvalue))]
    (log/debug "Mapper input: " (pr-str key)
               " => " (log/logstr value))
    (binding [*reporter* reporter]
      (doseq [[key value] (map-fn key value)]
        (log/debug "Mapper OUTPUT: " (pr-str key)
                   " => " (log/logstr value))
        (.collect output (Text. (pr-str key))
                  (Text. (pr-str value)))))))

(defn standard-reduce [reduce-fn this wkey wvalues-iter output reporter]
  (let [key (read-string (str wkey))
        values (map #(read-string (str %)) (iterator-seq wvalues-iter))]
    (log/debug "Reducer input: " (pr-str key)
               " => " (log/logstr values))
    (binding [*reporter* reporter]
      (doseq [[key value] (reduce-fn key values)]
        (log/debug "Reducer OUTPUT: " (pr-str key)
                   " => " (log/logstr value))
        (.collect output (Text. (pr-str key))
