 
 ;; ## Key-Value
 
+(defn kv-spec
+  "Returns a DomainSpec initialized with a hash-mod sharding scheme, a
+  key-value persistence and the supplied sharding count."
+  [shard-count]
+  (DomainSpec. (JavaBerkDB.)
+               (HashModScheme.)
+               shard-count))
+
 (defn hfs-tap
   "Returns an Hfs tap with the default sequencefile scheme and the
   supplied fields."

          data2 {0 (barr 1)
                 10 (barr 100)}]
      (fill-domain sink data)
      (check-results tmp data)
      (fill-domain sink data2)
      (check-results tmp (conj data2 [(barr 1) nil])))))

;; TODO: Invalid. Doesn't belong in this project; this makes far too
;; many assumptions about a thrift interface, etc. All we're concerned

    ret))

(def defaults
  {"io.serializations"
   (s/join "," ["org.apache.hadoop.io.serializer.WritableSerialization"
                "cascading.tuple.hadoop.BytesSerialization"
                "cascading.tuple.hadoop.TupleSerialization"])})

(def mk-props
  (partial merge defaults))

(defn job-conf
  "Returns a JobConf instance, optionally augmented by the supplied
   property map."
  ([] (job-conf {}))
  ([prop-map]
     (let [conf (JobConf.)]
       (doseq [[k v] (mk-props prop-map)]
         (.set conf k v))
       conf)))

(defn flow-connector
  "Returns an instance of FlowConnection, optionally augmented by the
   supplied property map."
  ([] (flow-connector {}))
  ([prop-map]
     (FlowConnector. (mk-props prop-map))))

(defn conj-serialization!
  "Appends the supplied serialization to the supplied JobConf
  object. Returns the modified JobConf object."
  [conf serialization]
  (let [old-val (.get conf "io.serializations")
        new-val (str old-val "," serialization)]
    (.set conf "io.serializations" new-val)))

(defn populate!
  "Accepts a key-value tap, a sequence of key-value pairs (and an
