@lyku/para-pipeline

import pipeline from "@lyku/para-pipeline";

@lyku/para-pipeline is a small streaming-iterator toolkit shaped like RxJS / IxJS but specialized for typed arrays. The win is fusion: a chain of @lyku/para-simd kernels (mulScalar, add, relu, …) collapses into a single pass at .run() time, so the intermediate arrays don’t get allocated. If the input is large enough that GPU dispatch wins (gpu.winsForSize(...)), the fused chain runs as one parabun:gpu simdMap instead.

Stage operators

Each operator is a transducer — a function Iterable → Iterable (sync or async). Chain with pipe or call them positionally.

Operator	Description
map(fn)	(x, i) => y.
filter(fn)	Keep when fn(x, i) is truthy.
take(n) / drop(n)	Window the iteration.
takeWhile(fn) / dropWhile(fn)	Window by predicate.
flat() / flatMap(fn)	Flatten one level / map+flatten.
chunk(n)	Non-overlapping arrays of size n (last group may be short).
windowed(n, step?)	Sliding window of size n, advancing by step (default 1).
pairwise()	Yields [prev, curr] tuples.
enumerate()	Yields [index, value].
scan(fn, init)	Running fold — emits each intermediate accumulator.
distinct(keyFn?)	Drop repeats anywhere in the stream.
distinctUntilChanged(eqFn?)	Drop adjacent repeats only.
tap(fn)	Side effect; passes items through.
delay(ms)	Sleep ms between yields.
throttle(ms)	Emit at most once per ms window.
debounce(ms)	Emit only after ms of upstream silence.
catchError(handler)	Recover from upstream errors with a value or substitute stream.
retry(times)	Restart the source on error up to times times.

Sinks

Sink	Description
collect()	Materialize as a plain Array.
toFloat32Array() / toFloat64Array()	Materialize as a typed array.
reduce(fn, init)	Fold.
forEach(fn)	Side-effect-only consumer.
count()	Count items.
sum()	Numeric sum (Kahan-compensated).
first(pred?) / last(pred?) / find(pred)	Selector terminals.
min(keyFn?) / max(keyFn?)	Extreme by numeric key.
topK(k, keyFn?, {by})	The k best, best→worst. Streaming bounded heap — O(n log k) time, O(k) memory, never sorts/buffers the dataset. by:"max" (default)/"min", stable on ties.
argTopK(k, keyFn?, {by})	Same, returns a Uint32Array of source indices (the “top rows” form).
every(pred) / some(pred)	Universal / existential.
toMap(keyFn, valueFn?) / toSet	Collect into Map / Set.
groupBy(keyFn)	Map<K, T[]>.
partition(pred)	[matched[], unmatched[]].

Sources

range(start, end, step?)

Lazy range generator. Useful as a chain head when you want a numeric stream without materializing.

import pipeline from "@lyku/para-pipeline";

const evenSquares = pipeline.range(0, 1_000)
  .filter(x => x % 2 === 0)
  .map(x => x * x)
  .toFloat32Array();

Other sources

of(...values)	Wrap arguments as a one-shot iterable.
from(source)	Identity wrapper — handy in dynamic chains.
empty()	Yields nothing.
concat(...sources)	Run sources end-to-end.
merge(...sources)	Race-style interleave across async sources.
zip(...sources)	Lockstep tuples; stops at the shortest source.
repeat(source, n?)	Replay a source n times (default infinite).

Bring your own

Any iterable / async iterable works as a source — typed arrays, @lyku/para-csv row streams, parabun:audio capture frames, anything.

for (const piece of pipeline.range(0, 1000).filter(x => x % 2).map(x => x * x).chunk(100)) {
  process(piece);
}

pipe(source, ...stages)

Compose without method-chain awareness — useful when stages are passed dynamically:

import { pipe, map, filter, sum } from "@lyku/para-pipeline";

const total = pipe(
  data,
  filter(x => x > 0),
  map(x => x * 2),
  sum(),
);

pipeParallel(source, ...stages)

Same shape, but the iterable is consumed across @lyku/para-parallel’s worker pool. Each worker processes a chunk through the entire stage chain, then results are merged. Stages must be pure (same constraint as pmap).

Fusion + GPU lift

When every stage in a chain is a @lyku/para-simd kernel (the documented set: mulScalar, addScalar, add, mul, Math.* body via simdMap), the call to .toFloat32Array() walks the chain and emits a single simdMap call covering the composed function. No intermediates allocated.

If gpu.winsForSize returns true at the chain’s input size, the fused chain runs on GPU instead of CPU SIMD — same call site, dispatched.

const ys = pipeline.range(0, 1_000_000)
  .map(x => x * 2)
  .map(x => x + 1)
  .map(x => Math.sqrt(x))
  .toFloat32Array();
// → single GPU simdMap kernel: x => Math.sqrt(x * 2 + 1)

Top-K over large / sharded data

source |> sort() |> take(k) is not a sort — it’s selection. topK does it in one streaming pass with an O(k) heap; the dataset is never sorted or materialized. Combined with the columnar projection sources you get “top rows by score over an arbitrarily large CSV” at O(batchSize + k) memory — the parser holds one batch, the heap holds k:

import csv from "@lyku/para-csv";
import p from "@lyku/para-pipeline";

const top5 = await p.topK(5, r => r.score)(
  p.fromColumns(csv.parseBatches(file, { schema: { id: "string", score: "f32" }, batchSize: 8192 }), ["id", "score"]),
);

fromColumn(batches, name) / fromColumns(batches, names) project a column-batch stream to per-row scalars / a per-row object of just those fields — no full-row objects. They are structural: both the @lyku/para-csv parseBatches shape and an @lyku/para-arrow RecordBatch work; this module depends on neither.

topK is a monoid — mergeTopK([topK(A), topK(B)], k) ≡ topK(A ∪ B) — so multi-file / multi-shard top-k is local-top-k per shard then a synchronous merge, at O(shards·k) memory regardless of total rows:

const locals = await Promise.all(files.map(f => p.topK(5, keyFn)(streamOf(f))));
const global = p.mergeTopK(locals, 5, keyFn);

Limits

• Fusion only collapses arithmetic + Math.* + ternary bodies. Branchy or stateful operators (filter, chunk, anything that breaks the 1-in-1-out shape) act as fusion barriers.
• pipeParallel adds the worker-pool overhead — see @lyku/para-parallel for when that pays off.
• The chain executes lazily — operators don’t run until a sink pulls. If you tap(console.log) and never call a sink, nothing prints.