   optional JobConf instance, supplied with the :conf keyword argument)
   and sinks all key-value pairs into the tap. Returns the original tap
   instance.."
-  [kv-tap kv-pairs & {:keys [props]}]
-  (with-open [collector (.openForWrite kv-tap (job-conf props))]
+  [kv-tap kv-pairs]
+  (with-open [collector (.openForWrite kv-tap (job-conf))]
     (doseq [[k v] kv-pairs]
       (.add collector (Tuple. (into-array Object [k v])))))
   kv-tap)

            ElephantDBTap$Args KeyValTailAssembly KeyValGateway]
           [org.apache.hadoop.mapred JobConf]))

;; ## Key-Value

(defn kv-spec


(defn connect!
  "Connect the supplied source and sink with the supplied pipe."
  [pipe source sink]
  (doto (.connect (flow-connector) source sink pipe)
    (.complete)))

(defn tuple-seq
