 (defn ingest-labels-from-path!
   [graph path]
   (doseq [file (kafka/dir->files path)]
-    (let [filename (.getName file)
-          element (string/lower-case (kafka/path->label filename))
-          lines (line-seq (io/reader file))
-          parsed (map #(json/parse-string % keyword) lines)]
-      (ingest-label-collections! graph element parsed filename))))
+    (try
+      (let [filename (.getName file)
+            element (string/lower-case (kafka/path->label filename))
+            lines (line-seq (io/reader file))
+            parsed (map #(json/parse-string % keyword) lines)]
+        (ingest-label-collections! graph element parsed filename))
+      (catch Exception e
+        (log/info "file failed to parse: " (.getName file))))))
 
 (def parse-args
   [["-c" "--config CONFIG" "path to config file"]

   :order order
   :offset offset
   :count qount
   :sort sortBy
   :groupCount group-count})

(defn apply-step

  [direction]
  (if (= "desc" (name direction)) -1 1))

(defn psort
  [field direction]
  (let [order (sort-order direction)]
    [{:$sort {(name field) order}}]))
