   {"insert"  insert-file
    "extract" extract-to-csv})
 
-; time ls ../data/*.gz | grep -E '(yellow|green)' | shuf | xargs java -Xms16g -Xmx30g -jar target/taxi-rides-clj-0.0.1-SNAPSHOT-standalone.jar insert  6
+; time ls ../data/*.gz | grep -E '(yellow|green)' | shuf | xargs java -Xms16g -Xmx30g -jar target/taxi-rides-clj-0.0.1-SNAPSHOT-standalone.jar insert  4
 ; time ls ../data/*.gz | grep -E '(yellow|green)' | shuf | xargs java -Xms16g -Xmx30g -jar target/taxi-rides-clj-0.0.1-SNAPSHOT-standalone.jar extract 8
 (defn -main [f-name n-parallel & files]
   (do
     (my-println "Started!")
     (create-taxicab-template)
     (doall (cp/upfor (Integer. ^String n-parallel) [file files] (->> (string/replace file #".+/" "") (str data-folder) ((main-funs f-name)))))
-    (my-println "Done, filtered " @collision-counter " duplicate documents")
+  ; (my-println "Done, filtered " @collision-counter " duplicate documents")
     (shutdown-agents)
     (System/exit 0)))

                     ;:_source {:enabled false}
                      :properties (->> (map vector (keys fields) (map es-types (vals fields)))
                                       (into {}))}}
   :settings {"index.translog.flush_threshold_size" "1g"
              "index.store.throttle.max_bytes_per_sec" "1000mb"
              :number_of_shards 1
              :number_of_replicas 0
              :refresh_interval "10s"}})
;(pprint taxi-mapping)

(def es-config

                                  (do 
                                    (my-println-f "Exception " (.getMessage e) ", retrying after " retry-t  " ms...")
                                    (Thread/sleep retry-t))))))
          process-chunk (fn [chunk-in]
                          (let [chunk-body (->> chunk-in (map doc->bulk) flatten s/chunks->body)]
                          ; Call insert-chunk until it returns non-nil. I was going to recur
                          ; but it could not be done from "catch" position so this is what I came up with
                            (->> #(insert-chunk chunk-body) repeatedly (filter some?) first)))
          _       (my-println-f "started")
          freqs   (->> (parse-trips file in) (partition-all 1000) (map process-chunk) flatten frequencies)
          _       (my-println-f "got " freqs)]
        freqs)))

