    :travel-h     [:float    :travel-h]
    :speed-kmph   [:float    :speed-kmph]})
 
+; Mapping CSV columns to lowercase to simplify schema definition
+(def col-mapping (into {} (for [[k v] col-mapping-raw] [(if (string? k) (string/lower-case k) k) v])))
+
 (def fields
   (->> (concat
          (for [weather-field (-> weather vals first keys)] [weather-field :float])

(defn parse-trips [file]
  (with-open [in (-> file clojure.java.io/input-stream java.util.zip.GZIPInputStream.)]
    (let [doc-id-fields [:pickup-dt :dropoff-dt :pickup-lat :pickup-lon :dropoff-lat :dropoff-lon :travel-km :paid-total]
          dataset (->> file str (re-find #"([a-z]+)[^/]+$") second)
          [header & rows] (-> in slurp csv/read-csv)
          header   (map string/trim header)
          missing  (clojure.set/difference (->> col-mapping keys (filter string?) set) (-> header set))
          check    (if-not (empty? missing) (throw (Exception. (str "Missing cols for " missing " at " file))))
          drop-fields identity ; #(reduce dissoc % [:pickup-lat :pickup-lon :dropoff-lat :dropoff-lon])
          header   (map col-mapping header)
          n-cols   (count header)
          round    (fn [n] (let [scale 1000.0] (-> n (* scale) Math/round (/ scale))))
          to-int-str (fn [v] (if v (let [v (int v)] (str (if (< v 10) "0" "") v))))
          deg2km   (/ 40075.0 360.0)
          mile2km  (fn [mile] (-> mile (* 1.60934) round))
          delta-km (fn [from to] (if (and from to) (-> (- to from) (* deg2km) round)))
          dist     (fn [a b] (if (and a b) (Math/sqrt (+ (* a a) (* b b))) 0.0))
          to-time  (fn [datetime]
                     (let [time (apply + (map / (map #(->> % (apply str) ((:int parsers))) (->> datetime (drop 9) (partition 2))) [1.0 60.0 3600.0]))]
                       (if (not= time 0.0) (min (round time) 23.999))))
          docs     (for [row rows :when (>= (count row) n-cols)]
                     (let [doc (into (sorted-map) (map (fn [h v] (if h [(second h) ((-> h first parsers) v)])) header row))]

              (assoc  :dlon-km       (delta-km (:pickup-lon doc) (:dropoff-lon doc)))
              (assoc  :pickup-time   (to-time (:pickup-dt  doc)))
              (assoc  :dropoff-time  (to-time (:dropoff-dt doc)))
              (update :paid-ehail    coalesce-to-zero)
              (update :paid-tax      coalesce-to-zero)
              assoc-travel-h
              assoc-speed-kmph
              filter-km-deltas
              drop-extra-fields)))))

;(time (->> #"green_.+\.csv\.gz$" find-files second parse-trips count))
;(->> #"green_.+\.csv\.gz$" find-files first parse-trips (take 3) pprint)
;(->> #"yellow_tripdata_2009-07\.csv\.gz$" find-files first parse-trips (take 3) pprint)
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(def es-types
