           " from " (:request_uri download))))
 
 
-(defn my-map [download]
+(defn my-map [key download]
   (status-code-dispatch (decode-response-body download)))
 
 
 
 ;;; MAPPER METHODS
 
-(defn load-jruby []
-  (alter-var-root #'*ruby*
-                  (fn [_] (.getEngineByName (ScriptEngineManager.) "jruby")))
-  (alter-var-root #'*ruby-context*
-                  (fn [_] (.getContext *ruby*)))
-  (.eval *ruby* (str "require 'org/altlaw/extract/scrape/run_scraper'"))
-  (.eval *ruby* (str "$runner = RunScraper.new")))
+(defn mapper-configure [this job]
+  (context/use-hadoop-jobconf job))
 
-(defn mapper-configure [this job]
-  (let [cached (DistributedCache/getLocalCacheFiles job)
-        path (first (filter #(= (.getName %) "docids.tsv.gz") cached))]
-    (when (nil? path) (throw (RuntimeException. "No docid.tsv.gz in DistributedCache.")))
-    (info "Loading docid map from " path)
-    (alter-var-root #'*docid-map*
-                    (fn [_] (tsv/load-tsv-map (str path))))
-    (load-jruby)))
-
-(defn mapper-map [this wkey wvalue output reporter]
-    (let [input (walk/keywordize-keys (read-json-string (str wvalue)))]
-      (debug "Called map with file position " wkey
-             " and value " (logstr input))
-      (binding [*reporter* reporter
-                *raw-input* (str wvalue)]
-        (doseq [[key value] (my-map input)]
-          (.collect output (Text. (str key)) (Text. value))))))
+(def mapper-map (partial h/json-map my-map))
 
 
 

;;; TOOL METHODS

(defn tool-run [this args]
  (let [job (hadoop/default-jobconf this)
        outpath (Path. (hadoop/job-path :pre :altcrawl))
        hdfs (FileSystem/get job)]
    (.setJobName job "pre.altcrawl")
    (FileInputFormat/setInputPaths job (str (hadoop/job-path :seqfiles :altcrawl) "/*.jsr"))
    (FileOutputFormat/setOutputPath job outpath)
    (.setInputFormat job TextInputFormat)
    (.delete hdfs outpath true)
    (.setMapperClass job (Class/forName "org.altlaw.jobs.pre.altcrawl_mapper"))
    (.setReducerClass job (Class/forName "org.altlaw.jobs.pre.altcrawl_reducer"))
    (.setMapOutputKeyClass job Text)
    (DistributedCache/addCacheFile (URI. (hadoop/docid-path :seqfiles :ohm1)) job)
    (JobClient/runJob job))
  0)


;; Step 3 (for HTML): Run scrapers

(defn handle-scraper-record [record]
  [101 (pr-str (into {} record))])

(defn run-scrapers [download]
  (map handle-scraper-record
       (ruby/eval-jruby "$handler.parse(Download.from_map($dmap))"
                        {:dmap (walk/stringify-keys download)
                         :handler (scraper-handler)})))


;;; Step 4 (for PDFs): convert to HTML
