clojure-hadoop shuffle component example

gistfile1.clj

;; Pseudocode example for guards in flows

(def *guard-time* [:days 7])

(defmacro with-resource-time-guard 
  [[name timespec inhibit?] &body ]
  `(let [resource# (get-resource name)]
     (cond (and ~inhibit? (resource-value resource#))
           (resource-value resource#)
           (or (resource-expired? resource# ~timespec)
               (nil? (resource-value resource#)))
           (let [new-value# (do ~@body)]
             (set-resource! resource# new-value#)
             new-value))))

(defmacro with-output-file 
  [[var output-file] & body]
  (assert (symbol? var))
  `(let [~var ~output-file]
      ~@body
      ~var))

(define-flow guarded-file-preprocessor [table outputBase inhibit-guard?]
  (with-resource-time-guard ["MyResource" *guard-time* inhibit-guard?]
    (with-output-file [dir (make-output-dir outputBase)]
      (do-step file-preprocessor table dir))))

(define-flow major-flow [table output]
  (do-step kmeans (guarded-file-preprocessor table *output-base* nil) output))