Processing pipelines: threading, for, transducers

Real programs spend most of their time turning one shape of data into another: log lines into a dashboard, rows into a report, events into a running total. Clojure gives you three idiomatic ways to express that transformation — -‌>‌>clojure.core/->>Thread the value through forms by inserting it as the last arg.view on clojuredocs → threading, sequence comprehensions, and transducers — and each fits a different situation. The choice is fundamentally imperative vs declarative: describe the transformation as data flow, not as mutable steps. Knowing which form to reach for is the practical skill the standard library hides behind its eighty-something seq functions.

The default: thread with -‌>‌>clojure.core/->>Thread the value through forms by inserting it as the last arg.view on clojuredocs →

The -‌>‌>clojure.core/->>Thread the value through forms by inserting it as the last arg.view on clojuredocs → macro reads top-to-bottom, each step takes the previous step's output as its last argument, and the whole pipeline is a single expression. For most data work this is the clearest possible form:

You can read it as a recipe: take this collection, keep the hits, pull out the latency, sum them up. No intermediate names, no loops, no off-by-one.

The cost: each stage walks the collection and allocates an intermediate seq. For five elements that's free; for fifty million it isn't.

Comprehensions when the shape is "for each X, for each Y"

When the transformation is most naturally read as nested iteration with a filter, f‌o‌rclojure.core/forList comprehension; lazy. (for [x xs :when (p x)] (f x)).view on clojuredocs → makes the structure explicit:

A two-binding f‌o‌rclojure.core/forList comprehension; lazy. (for [x xs :when (p x)] (f x)).view on clojuredocs → with :when is often clearer than mapcat + f‌i‌l‌t‌e‌rclojure.core/filterLazy seq of items where (pred x) is truthy.view on clojuredocs →, and :let lets you compute a value once and use it in both the predicate and the body. Use it when the loop structure is the explanation.

Transducers when the source is huge or unknown

A transducer is a transformation factored out from the collection it runs on. The same (comp (filter …) (map …) (partition-by …)) value can run once over a vector, again over a channel, and once more over a lazy seq, and no intermediate seqs are ever materialised. This is the right tool when the pipeline is hot, the data is large, or the source isn't a collection at all.

Notice the pipeline (hit-latency) was built without mentioning the data. That same value works with i‌n‌t‌oclojure.core/intoPour all items of from-coll into to-coll.view on clojuredocs →, sequence, transduce, and core.async channels — the transducer is the what, the call site supplies the how.

The threading version and the transducer version compute the same answer; the transducer version walks the input exactly once and allocates nothing in between.

Choosing the form

A practical rule of thumb:

Don't reach for transducers prematurely. The threading form is easier to read and almost always fast enough; the transducer pays off when you can point at a benchmark or a memory profile.

A worked example: tally hits per route

A small log analysis pipeline. Pull HTTP-style entries, keep the successful ones, group by route, count each group, and return the top-3:

The same pipeline as a transducer, demonstrating the factoring:

Identical answer; the second version never allocates an intermediate seq of paths. Whether that matters depends on the input — the rule of thumb says "benchmark first."

In a Clojure service

Batch import endpoints are a natural fit for a -‌>‌>clojure.core/->>Thread the value through forms by inserting it as the last arg.view on clojuredocs → enrichment pipeline: each order in the request body passes through lookup, validation, and totalling as a single expression in the functional core. The handler stays thin — parse the body, call the pipeline, persist the enriched batch — while the pipeline itself is pure and testable without HTTP. When profiling shows the batch is large enough to matter, the same steps factor into a transducer without rewriting the logic.

(defn enrich-batch [db orders]                   ;; sketch
  (->> orders
       (map #(assoc % :customer (lookup-customer db (:customer-id %))))
       (map #(assoc % :fx-rate (lookup-fx db (:currency %))))
       (map compute-total)
       (filter valid-order?)))
 
(defn import-orders-handler [{:keys [db]} {:keys [body]}]
  {:status 200 :body (enrich-batch db (:orders body))})

Check yourself

Exercise

The imperative version below totals spend per customer, then keeps only customers over 100. Rewrite pipeline->map as a single -‌>‌>clojure.core/->>Thread the value through forms by inserting it as the last arg.view on clojuredocs → pipeline that returns the same map — no a‌t‌o‌mclojure.core/atomMutable, synchronous, uncoordinated reference type.view on clojuredocs →, no d‌o‌s‌e‌qclojure.core/doseqLike for, but for side effects; returns nil.view on clojuredocs →. Run until the assert passes.

(defn pipeline->map [events]
  (->> events
       (group-by :customer)
       (map (fn [[c es]] [c (reduce + (map :amount es))]))
       (filter (fn [[_ total]] (> total 100)))
       (into {})))