crowley: ripgrep for json

crowley is a tool for querying, tabulating, and profiling streaming JSON with O(buffer) memory, whatever the input size. You can:

• Answer many queries in a single pass
• Gather multiple outputs (value streams, aggregations, summaries) per query
• Turn arbitrarily nested, schemaless JSON into the tabular form of your choice bounded only by your output size
• Query many heterogenous data sources in parallel
• Run the same tool on a local machine or a cloud instance, on a fraction of a core or on many cores, on local, streamed, or compressed data, on boundedly-sized data or continuous web socket output.
• All with minimal memory usage and at 2+ GB/s on a single core: see detailed benchmarks in Performance.

Currently supports NEON, AVX-2, and AVX-512 instruction sets. While capable of falling back to a fully scalar implementation in the absence of supported instruction sets, it is considerably slower and not recommended for production use.

Installation

pip install pycrowley

The distribution is named pycrowley (PyPI reserves the bare name), but the package you import is plain crowley:

import crowley

Optional extras: pip install 'pycrowley[remote]' adds fsspec for s3:///https://… URL sources (add the scheme's backend too, e.g. s3fs); 'pycrowley[dataframe]' adds numpy + pandas for to_pandas() and real numeric dtypes. to_polars()/to_arrow()/write_parquet() use polars/pyarrow if you have them installed; nothing else is required.

Wheels cover CPython ≥ 3.9 on macOS (arm64, x86_64) and Linux (x86_64, aarch64, manylinux2014). No Windows wheel yet.

Quick Start

import crowley

# iterate lazily (à la ijson)
for event in crowley.items("events.json", "events[*]"):
    ...     # your logic goes here 


# extract a multi-column table in a single pass 
# works with arbitrarily-nested JSON, not just NDJSON!
df = crowley.table(
    "events.json", rows="events[*]",
    columns={"id": "id", "repo": "repo.name", "at": "created_at"},
).to_pandas()


# set up a store elsewhere, e.g. a list
unique_values = []
# append streamed values into the list *after* the run completes
crowley.Query("events.json", 
              "events[*].repo.name", 
              crowley.ValuesUnique(into=unique_values))

# or, set numeric=True to coerce matched values to ``float`` and append
# a single NumericBuffer to ``into`` instead. 
import numpy as np
out = []
crowley.Query("events.json",
              "events[*].created_at",
              crowley.Values(into=out, numeric=True))
timestamps = np.frombuffer(out[0].buffer, dtype=np.float64)
# if ``into`` is omitted from Values or ValuesUnique, they are equivalent
# to ``Count`` and ``CountUnique``, respectively


# run a query over an unbounded socket+pipe stream, progressively printing 
# the count of matching elements until a condition is met
def cb(bytes_scanned: int, results: list[crowley.QueryResult]):
    region_dict = results[0].value_counts()
    print(region_dict)      # e.g. { europe: 123, asia: 456,  ... }
    return region_dict.get("north_america", 0) > 1000
    # when we reach a certain condition, terminate gracefully
    # by returning a truthy value
    # otherwise, it will run until the input fails or the process aborts

sock = socket.create_connection(...)
# take the reader and make into a file-like object
q = crowley.Query(
    crowley.Source(sock.makefile("rb"), progress=cb, compression="none"), 
    "requests[*].region", 
    crowley.ValueCounts())
# final tally after scan finishes
q.result(0, 0).value_counts()


# make multiple queries and output multiple statistics in a single scan
q = crowley.Query("events.json", {
    "events[*].type": [crowley.ValueCounts()],
    "events[*].payload.size": [crowley.Sum(), crowley.Max()],
})
# grab the result of (source 0, query 0)
# and extract the value_counts result output (a dict)
result_dict = q.result(0, 0).value_counts()
# grab the query by name rather than index,
# and extract the sum result output (a numeric)
result_sum = q.result(0, "events[*].payload.size").sum()


# perform a query in parallel across many sources
# you can pass a glob, a list, a list with globs, a mixed
# list of files, globs, urls, inline JSON, and more!
q = crowley.Query("data/bug_reports/*.json",
                  "reports[*].severity",
                  crowley.ValueCounts())
# view the individual sources and results after globbing and deduplication
for s in q.sources: print(s, q.result(s, 0).value_counts()) 
# get the result from a particular source by name
q.result("data/bug_reports/july_2025.json", 0).value_counts()

Sources

Every entry point takes the same source forms:

• A path: "data.json". Will stream-decode natively when the extension is .gz, .zst, or .zstd.
• A glob: "logs/*.json", expanded with shell semantics and deduplicated against literal paths. NOTE: * doesn't match hidden dot-files.
• A URL: "s3://bucket/key.json.gz", "https://…" opened through fsspec (optional dependency, plus the scheme's backend such as s3fs). As with paths, compression is inferred from the extension.
• A file-like object: anything with read(size), such as an open file, an HTTP response body, a boto3 StreamingBody, or an fsspec file. crowley automatically handles closing file-like objects which it opened. Any file-like object which the user opened and passed to crowley is not closed automatically, and must be handled by the caller.
• Inline JSON: a str beginning with {/[, or bytes/bytearray.
• A configured source: crowley.Source(src, ...) wraps any of the above to attach options that belong to one source rather than to the run:
  • progress= / progress_interval=: a callback for this one source. See progress below.
  • name=: a label for result(name, ...), equivalent to the {name: source} dict form.
  • compression="gzip" | "zstd" | "none": overrides extension inference, necessary to decode a compressed file-like object, which has no extension to infer from. Wrapping a glob applies the override to every match.

A Source carrying a callback or a name must resolve to exactly one file: a glob fanning out under one name would make the name ambiguous, and one callback across several files would make its byte counts meaningless.

Query accepts a list of many sources of mixed types, scanned in parallel on a worker pool, each with its own results. Unbounded streams (sockets, pipes) mix in like any other source. Memory use remains bounded, and each stream can carry its own progress= callback to watch it and stop it (though accumulated results constructed from the input will grow unless stored elsewhere).

NDJSON and JSONL do not need separate modes: the scanner reads concatenated root values natively. You may query with paths relative to each root, or anchor a table with rows="" (the empty query, matching each root value). NOTE: Scalar values at root (outside an array) are not supported.

The query language

Queries are regular expressions over paths:

Operator	Example	Meaning
Sequence	foo.bar.baz	field foo, then bar, then baz
Field	foo / "foo bar"	exact field (quote for spaces/metachars)
Array index	[0], [1:3], [3:]	index or slice (exclusive end)
Wildcards	* / [*]	any one field / any one index
Disjunction	foo | bar	match either foo or bar
Optional	foo?.bar	with or without the foo step
Repetition	foo*	zero or more repeats of the step
Grouping	foo.(bar|baz).qux	nest freely
Kleene star	**	Match zero or more field accesses

Recursive descent can be constructed from the algebra: (* | [*])*.name matches name at any depth. The empty string "" is the empty path, the root value itself (each root, for NDJSON).

Quoted fields use JSON escapes, but a field name containing a character JSON itself must escape (e.g. a double quote, backslash, or control character) is rejected at parse: the engine matches keys by their raw document bytes, where such a character never appears unescaped, so the query cannot match.

Reach such keys structurally instead: a * wildcard, or kvitems on the enclosing object (whose yielded keys are properly unescaped).

Iteration: items and kvitems

for user in crowley.items( # lazily read a single source
    "data.json", 
    "users[*]", 
    accumulators=[crowley.ValuesUnique(), crowley.Types()]): # carry accumulators along
    ...
for key, val in crowley.kvitems("data.json", "config"):    # members of an object
    ...

Values are yielded as the source is read; abandoning the iterator abandons the scan. limit=n stops after n matches per query and stops reading. Multiple queries interleave through one iterator, and can be distinguished using with_index=True to get (query_index, value). it.stop(query_idx) silences one query mid-flight (already-buffered values still drain).

The scan ends only when all queries are stopped or satisfied.

In addition to this lazy iteration, you may use accumulators=[...] to also use crowley's regular accumulators, and read them with it.results().

Note that if it.results() is called before exhaustion the scan ends and the results up to that point are reported alongside bytes_covered.

Aggregation: Query

q = crowley.Query(sources, queries, accumulators, limit=None, mode="auto",
                  progress=None, progress_interval=None)
q.result(source, query)           # -> QueryResult; index or name, either slot

queries is a string, a list (sharing one accumulators list), or a {query: [accumulators]} dict. sources may be a {name: source} dict; result() then takes the name (unnamed paths answer to their own spelling, and any query answers to its text). q.sources / q.queries / q.accumulators list the valid addresses, and a wrong name raises KeyError naming them.

Accumulators, each read back from QueryResult via its accessor (an accessor whose accumulator wasn't requested returns None):

Accumulator	Read back with	Result
Count()	.count()	number of matched values
CountUnique()	.count_unique()	number of distinct matched values
ValueCounts()	.value_counts()	{value: count} dict
Exists()	.exists()	whether anything matched at all
Min()	.min()	minimum of numeric matches
Max()	.max()	maximum of numeric matches
Sum()	.sum()	sum of numeric matches
Mean()	.mean()	mean of numeric matches
Product()	.product()	product of numeric matches
Mode()	.mode()	most frequent value(s) as [value, count] pairs, ties included
Types()	.types()	set of JSON type names seen at matched positions
Values(into=…)	your into list + .count()	every matched value, streamed
ValuesUnique(into=…)	your into list + .count_unique()	distinct matched values, streamed

On the streaming pair, numeric=True fills a zero-copy float64 NumericBuffer instead of Python objects: use numpy.frombuffer(buffer, dtype=np.float64) to bring it into Python efficiently.

• limit=n bounds matches per query. When all queries are satisfied, the scan stops and bytes_covered records how far it got.
• progress=cb takes any callable and calls it as cb(bytes_scanned, [QueryResult, …]) roughly every progress_interval bytes (default 8MB; it can't fire more often than once per internal scan buffer, 1–4MB). Return True to stop the scan gracefully (results so far become final, with bytes_covered recording how far it got); raise to abort it. This allows us to deal with unbounded streams, and also allows progress bars to hook in. See the example with tqdm below:

from tqdm import tqdm

with tqdm(total=os.path.getsize(path), unit="B", unit_scale=True) as bar:
  def cb(bytes_scanned, results):
      bar.n = bytes_scanned
      bar.refresh()
  q = crowley.Query(path, "events[*].id", [crowley.Count()], progress=cb)

• Multiple sources and callbacks: when using multiple sources, attach each callback to its source with crowley.Source(src, progress=cb). The run-level progress= kwarg can only take a single source because a single callback over interleaved sources would have no meaningful byte count. A truthy return stops only that source and finalizes its results while other sources keep scanning, while an exception aborts the whole run. Callbacks run one at a time on the constructor thread in no defined order.

# watch two unbounded streams while a backfill file scans alongside
q = crowley.Query(
    [
        crowley.Source(us_sock.makefile("rb"), name="us", progress=us_cb),
        crowley.Source(eu_sock.makefile("rb"), name="eu", progress=eu_cb,
                    progress_interval=1_000_000),
        "backfill.ndjson"
    ],
    "requests[*].region", [crowley.ValueCounts()])
q.result("us", 0).value_counts()      # final tally per stream

• Scans release the GIL and are Ctrl-C interruptable to within ~50ms.

Tables

# create a table with the declared columns from a JSON stream
t = crowley.table(
    "events.ndjson", rows="",
    columns={"id": "id", "repo": "repo.name", "stars": "payload.stars",
             "at": "created_at"},
    # optional type declaration: nullable int32 and non-nullable timestamp
    dtypes={"stars": "int32?", "at": "timestamp"},
)
# turn a table into a dataframe...
t.to_pandas()
t.to_polars()
# or an in-memory pyarrow.Table (write_parquet writes parquet files directly)...
t.to_arrow()
# or a dictionary...
t.to_dict()
# or extract one column (numeric columns come back as NumPy arrays,
# others as plain lists)
t.column("id")

rows is the path to the container that contains a single row of data; columns maps output names to paths relative to the row (or is a list of paths used as names). Anchoring with rows ensures that a record missing a field gets a null in its own row, so column lengths remain consistent and the position of missing data isn't misattributed. A column matching more than once per row raises a ValueError unless the on_multi="first" | "last" | "list" kwarg is set. Fields which are not requested are never materialized.

Unless declared explicitly, data types are inferred per column (int64, float64, bool, str, nullable variants, object for a mix): int8/16/32/64, float32/64, bool, str, timestamp, uuid (yielding real uuid.UUIDs), decimal(p,s). Where a type is declared, crowley will throw an error rather than coerce. A missing field is a null under any declaration, but an explicit JSON null is refused by a plain declaration; append ? ("int32?", "decimal(10,2)?" etc) to declare the column nullable and accept explicit nulls as nulls. to_arrow() exports over the Arrow C Data Interface with zero copies for numeric columns (crowley itself carries no arrow dependency, though consuming the result requires pyarrow installed).

For sources bigger than memory, stream chunks instead of holding the table:

for chunk in crowley.table_batches(src, rows="", columns=cols, batch_rows=65536):
    consume(chunk)        # each chunk is a Table; nothing accumulates

Or write straight to disk through pyarrow's writers (installed separately):

crowley.write_parquet(src, "out.parquet", rows="", columns=cols, dtypes=d)
crowley.write_csv(src, "out.csv", rows="", columns=cols)

Both stream record batches, so peak memory is determined by batch size, not overall file/stream size.

Reshaping: write_ndjson and ndjson_lines

# normalize: nested JSON in, NDJSON out
crowley.write_ndjson("dump.json", "events.ndjson", rows="events[*]")

# slim + compress: keep desired fields and debloat
crowley.write_ndjson("dump.json", "slim.ndjson.zst", rows="events[*]",
                     fields={"id": "id", "repo": "repo.name", "at": "created_at"})

# sink-agnostic: lazy bytes lines for S3 multipart, sockets, anything
for line in crowley.ndjson_lines("dump.json", "events[*]", fields=["id"]):
    upload(line)

One line per rows-matched record, streaming, memory bounded regardless of source size. Without fields=, each record re-emits whole, preserving member order. With fields=, each record becomes a flat object: missing fields are omitted, explicit nulls are kept. on_multi/limit behave the same as with tables. Output paths compress by extension (.gz, .zst, .zstd).

With the exception of newlines, the re-emitted values are byte-faithful to the original. Compare to jq, which routes a 20-digit id through a double, corrupting it, while crowley preserves it exactly.

Benchmarking re-emission on the 502MB nested JSON corpus:

Tool	Runtime (s)	Peak RSS (MB)
jq -c	15.1	2300
ijson (hand-rolled loop)	4.4	40
crowley	0.65	34

crowley completes the same re-emission task 23× faster than jq at 1/66th the memory, and completes a projection task at 0.35s/102MB.

Streaming group-by: group_by

t = crowley.group_by(
    "events.ndjson", rows="",
    by={"repo": "repo.name"},                 # 1+ key paths (str/list/dict)
    agg={
        "events": "count",                    # rows per group; no path
        "stars":  ("sum", "payload.stars"),
        "actors": ("count_unique", "actor.login"),
        "seen":   ("first", "created_at"),
    },
)   # -> Table: key columns + agg columns, one row per group

SQL's GROUP BY over the streamed rows in one pass: rows anchors what a record is (exactly as for tables), keys and agg paths are relative to it.

Supported operations: count, sum, min, max, mean, count_unique, first, last.

Groups come out in first-seen document order; a record missing a key field lands in the null group rather than being dropped. Missing and null values are skipped by every aggregate (as SQL), but a non-null non-number under a numeric op is an error. All-integer columns use int64, with overflow errors rather than turning into a float. Even a single fractional value promotes the entire column to float64.

The result is an ordinary Table, with the same to_pandas()/to_polars()/ to_arrow()/to_dict() methods available. Peak RSS is defined by O(groups), not O(rows). To guard against excessive memory use, max_groups defaults to 1 million, raising an error if this is exceeded. When large group counts are intended, raise this value or pass None. count_unique additionally holds each group's distinct values.

Profiling: describe

print(crowley.describe("unknown.json", limit_bytes=50_000_000))

Profile structures without passing a query. This returns every path, how many values, which JSON types, approximate distinct counts (None past the
distinct_cap kwarg, which tells us there's more than we care to count), and nesting depth. A quick pass to examine or confirm the structure of a file.

ijson compatibility

import crowley.ijson as ijson
for record in ijson.items(f, "item"):      # ijson prefix syntax, unchanged
    ...

For ease of migration from ijson, we provide a drop-in replacement for ijson's items and kvitems methods. ijson prefix strings are translated into their equivalent crowley query, including item for array elements and "" for the root.

Numbers arrive as Decimal by default, as ijson's do. To restore floats, pass use_float=True.

If you use ijson.parse() instead, explore the native crowley.Query API above as a more efficient and expressive way to perform the same work.

Errors and partial results

Failures return a descriptive type, from which results can be salvaged:

• crowley.Incomplete: the source ended before the document did (truncation, a dropped connection). .recover() returns the per-query results for the bytes that arrived. (A Ctrl-C is not an Incomplete; it surfaces as an ordinary KeyboardInterrupt.)
• crowley.Corrupted: the JSON is not valid. .recover_corrupted() returns the per-query results accumulated before failure. This output is NOT guaranteed to be correct.

Both carry a bytes_covered property describing how far they got before failing. A QueryResult salvaged from either error has the is_partial property set to True.

Note that a scan that stopped early on purpose (such as by satisfying all limits) is not partial, as nothing failed.

Ordering: completion order

Streamed values (items, kvitems, Values(into=…)) arrive in completion order: a container is yielded at its closing brace, and with multiple queries, values are interleaved, grouped by scan buffer. Any single query's stream is in document order end to end; cross-query arrival order is not meaningful.

Use with_index=True rather than zipping streams positionally (or use table, whose cells land by row identity). Accumulators and tables are order-insensitive and unaffected. This is necessary to perform multi-query scans in a single pass with bounded memory.

Performance

While crowley can be used in a wide variety of environments for many tasks, it excels in running multi-GB scans with O(buffer) peak memory use.

Throughput. On a 502MB GHArchive corpus (M3 Max, warm cache), a single-query scan runs at 3.5 GB/s in the default two-thread pipelined mode and 2.6 GB/s on one thread (mode="batched").

Small machine meets big file. Extracting a five-column table out of a 502MB NDJSON file, running on AWS Lambda (128MB, arm64, Python 3.13, us-east-1) taking the best of 4 warm runs from each tool. All tools returned identical results (when they succeeded).

tool	128MB Lambda	peak memory	$ / 1k runs
crowley	29.3s	102MB	$0.049
msgspec (hand-chunked)	43.5s	102MB	$0.072
polars	✗ OOM	—	$0.069 (at 512MB)
pyarrow	✗ OOM	—	$0.110 (at 512MB)
pandas	✗ OOM	—	✗ OOM at 512MB too
ijson	✗ timeout (600s)	—	$0.197 (at 512MB)

At this scale, whole-file engines simply fail. ijson, our baseline comparison for streaming engines, can handle the space constraints but times out. The only other tool that finishes, msgspec, needs to have its input split into chunks by hand, and crowley outcompetes it 1.5x in both time and cost.

This is on NDJSON. On the equivalent nested document, crowley's numbers are unchanged — while hand-chunking (msgspec's lifeline) no longer works without line framing, and of the rest only ijson can process a nested document in bounded memory at all.

crowley is outcompeted by polars when RAM and cores are abundant. On a 26MB NDJSON file at 1769MB of Lambda memory, polars finishes twice as quickly as crowley. crowley is preferred when data originates in nested form, input size is large or unbounded, and memory/CPU is constrained.

Execution modes and hardware

The mode= kwarg defaults to "auto", which detects local hardware and chooses the execution mode for highest throughput. It reads cgroup quotas as well as Lambda memory size, so cloud containers get sensible defaults without explicit configuration.

You may also explicitly set the execution mode to "pipelined", which uses two threads per source for maximum throughput, or "batched", which uses a single thread. "batched" is superior when fewer than 2 cores are available, including for sub-vCPU deployments, or when you wish to optimize for cost.

The scanner's SIMD kernel is chosen at import: NEON on aarch64, AVX-512/AVX2 on x86-64. The active kernel can be examined at runtime with crowley._active_kernel().

If no supported instruction set is detected, crowley will raise a warning and proceed with a scalar fallback.

SVE2 and SSE4.2 kernels are still in development.