Data Operations¶

These steps reshape records without calling an LLM. They rename, drop, combine, split and merge — everything between the source and the sink that costs nothing but CPU.

They are ordinary transforms: put them anywhere in a pipeline, except Concat, which starts one.

At a glance¶

Map(fn)¶

Rewrite each record. The record you return replaces the one that came in.

from datafast import Map

step = Map(lambda r: {**r, "length": len(r["text"])})
print(list(step.process(iter([{"text": "hello"}]))))
# [{'text': 'hello', 'length': 5}]

Nothing is merged for you. Map(lambda r: {"length": len(r["text"])}) drops every other column, which is sometimes what you want. Spread the record with {**r, ...} to keep it.

FlatMap(fn)¶

Turn one record into many, or into none.

from datafast import FlatMap

step = FlatMap(lambda r: [{"q": q} for q in r["questions"]])
print(list(step.process(iter([{"questions": ["Q1", "Q2"]}]))))
# [{'q': 'Q1'}, {'q': 'Q2'}]

Returning [] drops the record. That makes FlatMap a filter that can also reshape.

AddUUID(column="id", overwrite=False)¶

Give each record a unique id, so records stay traceable after later steps reshape them.

from datafast import AddUUID

record, = AddUUID(column="example_id").process(iter([{"text": "hi"}]))
print(sorted(record))
# ['example_id', 'text']

The id is a uuid4 as a string. With overwrite=False a record that already has the column passes through untouched — including when its value is None, because the check is presence, not emptiness.

Filter(fn=None, where=None, keep=True)¶

Keep or drop records. Give it either a function or a where condition, never both.

from datafast import Filter

step = Filter(where={"score": {"$gte": 7}, "category": "science"})
print(list(step.process(iter([
    {"score": 9, "category": "science"},
    {"score": 3, "category": "science"},
]))))
# [{'score': 9, 'category': 'science'}]

Passing neither fn nor where raises ValueError, and so does passing both.

Conditions¶

In where, a bare value tests equality. A dict applies operators. Every column in a where, and every operator in one column's dict, must hold.

The length operators work on anything with a length — a string, a list, a dict. The string operators reject non-strings instead of failing.

Logical forms¶

$or and $and take a list of conditions and sit at the top of where, not under a column.

from datafast import Filter

step = Filter(where={"$or": [{"category": "science"}, {"score": {"$gte": 9}}]})
print(len(list(step.process(iter([
    {"category": "history", "score": 9},
    {"category": "history", "score": 2},
])))))
# 1

Dropping instead of keeping¶

keep=False inverts the whole condition: matching records are dropped.

from datafast import Filter

step = Filter(where={"quality": {"$lt": 3}}, keep=False)
print(list(step.process(iter([{"quality": 1}, {"quality": 5}]))))
# [{'quality': 5}]

Group(by, collect=None, output_column=None, agg=None, min_per_group=None, max_per_group=None)¶

Collapse the records sharing a key into one record — chunks back into a document, reviews into a product.

from datafast import Group

step = Group(by="doc_id", collect="chunk", agg={"n": "chunk:count"})
print(list(step.process(iter([
    {"doc_id": 1, "chunk": "a"},
    {"doc_id": 1, "chunk": "b"},
]))))
# [{'doc_id': 1, 'chunk_list': ['a', 'b'], 'n': 2}]

The output record holds only the by columns, the collected lists and the agg columns. Every other column is dropped. Name in collect or agg anything you need downstream.

An agg value is written "column:function", or "column:concat:separator" to choose the separator for concat (the default is a newline). The nine functions:

A bad spec is caught when you build the step, not when the pipeline runs: a missing : or an unknown function raises ValueError immediately.

Pair(n=2, strategy="random", within=None, across=None, output_format="columns", max_pairs=None, seed=None)¶

Combine records into pairs or larger tuples — for preference data, comparisons, or multi-chunk questions.

from datafast import Pair

step = Pair(n=2, strategy="sliding", within="doc_id")
print(list(step.process(iter([
    {"doc_id": 1, "chunk": "a"},
    {"doc_id": 1, "chunk": "b"},
]))))
# [{'chunk_1_doc_id': 1, 'chunk_1_chunk': 'a', 'chunk_2_doc_id': 1, 'chunk_2_chunk': 'b'}]

output_format="columns" prefixes every column with chunk_1_, chunk_2_ and so on. output_format="list" instead writes a chunks column holding the whole records, plus one {column}_list per column.

A group holding fewer than n records produces nothing. An invalid strategy, output_format, or an n below 2 raises ValueError when you build the step.

"random" requires max_pairs. Random draws have no natural end, so the cap is how many tuples you want; without it the step raises ValueError when you build it. Since "random" is the default strategy, a bare Pair() raises too. Draws are made with replacement, so the same tuple can come out twice — set seed if you want the same dataset again.

Concat(*sources)¶

Stack pipelines end to end: run each one, yield all its records, then move to the next.

from datafast import Concat, Source

step = Concat(Source.list([{"text": "a"}]), Source.list([{"text": "b"}]))
print(list(step.process(iter([]))))
# [{'text': 'a'}, {'text': 'b'}]

Concat reads only its own sources, so it counts as a source itself: it must be the first step of the pipeline, and compile() rejects it anywhere else. Records arriving from upstream are discarded, not passed through.

Each source must be a complete pipeline of its own — it starts with its own source and holds no sink. Building one with no sources raises ValueError.

Join(right, on, how="inner", suffixes=("_left", "_right"))¶

Merge two pipelines side by side on a shared key. The left side is the upstream pipeline; the right side is the one you pass in.

from datafast import Join, Source

right = Source.list([{"user_id": 1, "action": "click"}])
step = Join(right, on="user_id")
print(list(step.process(iter([{"user_id": 1, "name": "ada"}]))))
# [{'user_id': 1, 'name': 'ada', 'action': 'click'}]

Columns named on both sides get the suffixes; key columns never do. An unmatched record carries only its own columns — the missing side is absent, not filled with None.

Keys are matched exactly, and a record missing a key column joins on None, which matches every other record missing it. Both sides are held in memory, and a key present several times on both sides yields one record per combination.

Things worth knowing¶

• Group throws away the columns you did not name. Only by, collect and agg reach the output record.
• Pair(strategy="random") needs max_pairs, and so does a bare Pair(). Set seed too if you want the same dataset twice.
• $or and $and narrow their siblings. In {"$or": [...], "score": {"$gt": 5}} a record must match the $or and score above 5. Every key in a where dict must hold.
• A missing column is not an error in Filter. It reads as None, so it fails most conditions but satisfies $ne, $nin and $exists: False.
• Concat discards upstream records and must be the first step.
• Group, Pair, Concat and Join hold every record in memory while they work. Filter first when the dataset is large.
• compile() stops checking columns after Map, FlatMap, Group, Pair, Concat or Join, because their output shape is not knowable in advance. It does still check that by, within, across and on name columns that exist.

Where to go next¶

• Pipelines & execution — how these steps chain together.
• Source and Seed — the steps that come first.
• Key Concepts — the record → step → pipeline model.
• Full Glossary — what record, column, step and transform mean here.
• API reference — generated signatures for everything on this page.