Why sequence + mapcat silently turns your streaming pipeline into an eager buffer, and how IReduceInit fixes it
In the previous article, we built an elegant transducer pipeline for streaming large files: a with-open-xf transducer managing resource lifecycle inside the chain, mapcat expanding one file into N rows, and transduce consuming everything in O(1) memory. It works perfectly for a single source.
How to stream large files with O(1) memory, composable transducers, and automatic resource cleanup
You need to process large files from cloud storage. Each file has a header line followed by tens of thousands of data rows. The rows must be parsed, transformed, and written to a database in batches.
| (deftest partition-every-test | |
| (testing "Should partition based on elapsed time" | |
| (let [item-count 5 | |
| result (into [] | |
| (comp | |
| (map deref) | |
| (partition-every 100 :millis)) | |
| (repeatedly item-count #(future (Thread/sleep 50) 42))) | |
| flatten-result (flatten result) | |
| count-result (map count result)] |
| (deftest sliding-sort-test | |
| (testing "Should sort all the given element incrementaly" | |
| (let [xf (sliding-sort 10 <) | |
| input [5 6 7 2 4 0 8 3 1 9] | |
| output (into [] xf input)] | |
| (is (= output [0 1 2 3 4 5 6 7 8 9])))) | |
| (testing "Should sort the given element decrementaly in a sliding window of 5" | |
| (let [xf (sliding-sort 5 >) | |
| input [5 6 7 2 4 0 8 3 1 9] | |
| output (into [] xf input)] |