 (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"]

        "else" "$$PRUNE"}}}
     {:$replaceRoot {:newRoot "$_root"}}]))

(def steps
  {:fromEdge from-edge
   :toEdge to-edge

   :order order
   :offset offset
   :count qount
   :sort psort
   :gather gather
   :groupCount group-count})

(defn apply-step
