            [java.io InputStream OutputStream DataInputStream DataOutputStream File]
            [java.util UUID]
            [java.net URI]
-           [clojure.lang MapEntry ExceptionInfo]))
+           [clojure.lang MapEntry ExceptionInfo]
+           [org.joda.time DateTimeUtils DateTimeZone]
+           [org.joda.time.format DateTimeFormat DateTimeFormatter]))
 
 (def required-ns (atom #{}))
 

   (cond (or (symbol? arg) (string? arg)) (make-pipeline* arg [] {})
         (seq? arg) (make-pipeline* nil arg {})
         :else (make-pipeline* nil [] arg))))

(defn run-pipeline
  {:doc "Run the computation for a given pipeline or PCollection."


(def ^:dynamic *pipeline-builder-caller* "unknown")

(defn create-timestamp
  []
  (let [formatter (-> "MMddHHmmss"
                      (DateTimeFormat/forPattern)
                      (.withZone DateTimeZone/UTC))]
    (.print formatter (DateTimeUtils/currentTimeMillis))))

(defn job-name-template
  [tpl args]
  (-> tpl
      (str/replace #"%U" (or (System/getProperty "user.name") "nemo"))
      (str/replace #"%A" *pipeline-builder-caller*)
      (str/replace #"%T" (create-timestamp))
      (str/lower-case)
      (str/replace #"[^-a-z0-9]" "0")))

(defn make-pipeline*
  ([itf str-args kw-args]
   (let [atomic-args (into {} (map (fn [kv] (let [[k v] (str/split kv #"=" 2)]
                                              [(str/camel-case (str/replace k #"^--" "")) v]))
