Transducers (Clojure 1.7) are composable algorithmic transformations that are independent of where their input comes from and where their output goes: a collection, a stream, a channel. A reducing function is the kind of function given to reduce: (whatever, input -> whatever). A transducer is a function from one reducing function to another.
Using transducers
Most sequence functions have an arity without the collection argument that returns a transducer. This is not currying: (filter odd?), (map inc) and (take 5) each return a transducer. They compose with comp; the composition runs the steps left to right, in the same order as a ->> pipeline.
(def xf (comp (filter odd?) (map inc) (take 5)))
(transduce xf + (range 100)) ; reduce eagerly
(transduce xf + 100 (range 100)) ; with an initial value
(into [] xf (range 1000)) ; into a collection
(sequence xf (range 1000)) ; a sequence, computed incrementally
(def e (eduction xf (range 5))) ; applied each time it is reduced or iterated| Function | Does |
|---|---|
(transduce xform f coll), (transduce xform f init coll) | reduces eagerly with the transducer applied to f; the initial value is init or (f) |
(into to xform from) | puts the transformed items into a collection (using transients when possible) |
(sequence xform coll) | a sequence made incrementally: it consumes the input as needed and fully realises intermediate operations, which differs from chained lazy sequence functions |
(eduction xform* coll) | a reducible and iterable application, performed again on each reduce or iteration |
completing | adds a completion arity to a plain reducing function so that it can be used with transduce |
The functions that return a transducer when the collection is omitted are: map cat mapcat filter remove take take-while take-nth drop drop-while replace partition-by partition-all keep keep-indexed map-indexed distinct interpose dedupe random-sample. cat is itself a transducer. halt-when is another, which stops the whole process when a predicate matches.
Writing a transducer
A transducer takes a reducing function rf and returns a function of three arities:
| Arity | Name | Duty |
|---|---|---|
() | init | call the init arity of rf |
(result input) | step | a reduction step; call the step arity of rf zero or more times: filter calls it if the predicate is true, map exactly once, cat once per inner item |
(result) | completion | produce the final value and flush state; must call the completion arity of rf exactly once. partition-all flushes its pending items here |
(defn dedupe []
(fn [xf]
(let [prev (volatile! ::none)]
(fn
([] (xf))
([result] (xf result))
([result input]
(let [prior @prev]
(vreset! prev input)
(if (= prior input)
result
(xf result input))))))))Early termination and new processes
(reduced x)wraps a value to say that reduction should stop;reduced?tests for it;deref(or@) unwraps it; a step function that uses a nested reduce must check for reduced values and convey them (the reference points tocat).ensure-reducedandunreducedare helpers: the first wraps a value unless it is already reduced, the second unwraps a value that is.- A process that uses transducers must stop supplying input when a step returns a reduced value, unwrap it before completion, and call completion exactly once.
- A process must keep to itself the function it gets by applying a transducer: it may be stateful and is not safe to share across threads.
- State is normally kept in a
volatile!(changed withvreset!andvswap!), though an atom also works.
Not described
Known, not yet covered
- Reference: the reducers library (
clojure.core.reducers:fold,foldcat,reducer,folder) and transducers in core.async channels ((chan 1 xf)) are only named;reducers.adocandcore/reducers.cljwere downloaded but not summarised.
See also
- Sequences and laziness – the lazy-sequence model
- Concurrency – channels with transducers