LogReducer

Reduce gigabytes of logs to a small, representative sample - keeping the patterns and anomalies that matter and dropping the repetition that does not. Memory-safe streaming, temporal awareness, and ML-based anomaly detection.

LogReducer is two tools in one package:

• A CLI. logreducer app.log reduces a file - or a SQL, ClickHouse, or Kafka source - straight from the shell. No code to write.
• A library with an IO-agnostic core. The engine has zero IO dependencies and reduces any re-iterable stream of str lines (a Source). Embed it in your own pipeline; the engine never manages the connection.

Features

• Memory-safe streaming: near-constant memory on multi-GB inputs; low, container-friendly defaults
• Four reduction modes: pattern (Drain3), anomaly (Isolation Forest), temporal, and hybrid
• IO-agnostic core: reduce a file, a list[str], a database cursor, or a Kafka stream through one Source seam
• Optional adapters: SQL (SQLAlchemy), ClickHouse (clickhouse-connect), Kafka (confluent-kafka) - install only what you use
• Engine-side sampling: seeded, dialect-aware SQL sampling and native TABLESAMPLE for cheap reduction of huge tables
• Embeddable: injection seams for a host application's own config cascade and logging standard

Installation

As a CLI tool

Install it as an isolated tool so its dependencies never clash with your other Python projects (needs Python 3.12+):

uv tool install logreducer     # recommended (uv)
# or
pipx install logreducer        # recommended (pipx)

To bundle an adapter extra with the tool: uv tool install "logreducer[clickhouse]".

As a library

uv add logreducer
# or
pip install logreducer

Optional extras (install only what you need):

uv add "logreducer[enhanced]"    # fuzzy dedup, faster hashing
uv add "logreducer[sql]"         # SQLSource (SQLAlchemy) - bring your own DBAPI driver
uv add "logreducer[clickhouse]"  # ClickHouseSource (clickhouse-connect)
uv add "logreducer[kafka]"       # KafkaSource / KafkaSink (confluent-kafka)

The logreducer command works under any install method (project venv, pip install --user, pipx/uv tool, system-wide). pipx / uv tool is the recommendation for end users - isolation without a manual venv.

Quick Start

Command line

# Reduce a file to stdout, or to a file with -o
logreducer app.log
logreducer app.log -o reduced.log -l enhanced -m hybrid

# JSON output, with run stats on stderr
logreducer app.log --format json -o result.json --stats

# Cap memory, estimate first
logreducer huge.log --max-memory 2 --estimate

Library

from logreducer import LogReducer

reducer = LogReducer(level="standard")

# Reduce a file (writes reduced.log + reduced.meta.json)
reduced = reducer.process_file("app.log", output_file="reduced.log")
print(f"{len(reduced)} representative lines")

# Reduce any re-iterable of lines - no file needed
lines = ["ERROR timeout upstream=payments", "INFO ok", "ERROR timeout upstream=payments"]
reduced = reducer.reduce(lines)

reduce() returns the reduced lines in memory. Pass return_metadata=True for a dict of {"lines", "stats", "config"}.

How it works

Everything streams: lines flow through dedup and template mining as one generator pipeline, so the unique-line set is never materialised. Memory stays near-constant regardless of input size.

flowchart LR
    subgraph sources ["Source (re-iterable str lines)"]
        direction TB
        F[file]
        L["list[str]"]
        Q[SQL / ClickHouse query]
        K[Kafka topic]
    end
    sources --> D["exact dedup<br/>(bounded hash cache)"]
    D --> FZ["fuzzy dedup<br/>(MinHash LSH, enhanced+)"]
    FZ --> DR["Drain3 template mining<br/>(LRU-bounded clusters)"]
    DR --> P["rank patterns<br/>(CRITICAL > ERROR > WARN)"]
    P --> O["top patterns<br/>x examples"]
    D -. "anomaly / hybrid" .-> A["TF-IDF + Isolation Forest<br/>(batch, capped by anomaly_max_rows)"]
    A --> O
    O --> out["stdout / file + meta.json / Sink"]

The anomaly branch is the one part that cannot stream (Isolation Forest is batch ML); cap its input with anomaly_max_rows when reducing very high-cardinality sources.

Choosing a mode and level

flowchart TD
    Q{"What matters<br/>in the output?"} -->|"recurring structure,<br/>smallest output"| P["mode=pattern<br/>(default, fastest)"]
    Q -->|"rare or unusual<br/>events"| A["mode=anomaly"]
    Q -->|"bursts and<br/>time windows"| T["mode=temporal"]
    Q -->|"unsure /<br/>maximum coverage"| H["mode=hybrid"]
    A -.-> N1["needs >= 10 unique lines,<br/>else everything passes through"]
    P -.-> N2["fuzzy dedup joins in<br/>at enhanced/maximum"]

Mode	Description	Best for
pattern	Drain3 template mining	Structured / application logs (fastest)
anomaly	Isolation Forest outlier detection	Security and error logs
temporal	Time-aware pattern analysis	Time-series and monitoring logs
hybrid	Pattern + anomaly combined	Maximum coverage

Level	Speed	Memory cap	Features
standard	Fast	0.5 GB	Deduplication + pattern extraction
enhanced	Moderate	1 GB	+ fuzzy dedup
maximum	Thorough	2 GB	+ wider pattern budget

Reducing from a database or Kafka

The CLI dispatches on the --dsn scheme; a library caller constructs the source directly.

# PostgreSQL / MySQL / SQLite via SQLAlchemy (needs logreducer[sql] + a driver)
logreducer --dsn postgresql://user@host/db --query "SELECT message FROM logs"

# ClickHouse via the native driver (needs logreducer[clickhouse])
logreducer --dsn clickhouse://user@host:8123/db --query "SELECT message FROM logs"

# Kafka topic (needs logreducer[kafka])
logreducer --dsn kafka://broker:9092 --topic app-logs --group logreducer

from logreducer import LogReducer
from logreducer.clickhouse import ClickHouseSource

reducer = LogReducer(level="enhanced", mode="hybrid")
with ClickHouseSource("clickhouse://user@host:8123/db", "SELECT message FROM logs") as source:
    reduced = reducer.reduce(source)

The query selects the log line as its first column. Sources are re-iterable (the engine makes multiple passes), so a database source re-runs its query per pass and a Kafka source re-reads from the earliest offset without committing.

Sampling large sources

For a table too large to scan in full, sample a fraction of rows. SQL sampling is deterministic when you pass a seed (so the reducer's multi-pass modes see a stable input); ClickHouse uses its native SAMPLE clause.

from logreducer.sql import SQLSource

# Wrap any query in a seeded random predicate (PostgreSQL/MySQL)
source = SQLSource("postgresql://user@host/db", "SELECT msg FROM logs", sample=0.01, sample_seed=42)

# Or sample a TABLE with the engine's native sampler - PostgreSQL
# TABLESAMPLE SYSTEM is page-level and genuinely sub-linear:
source = SQLSource.from_table("postgresql://user@host/db", "logs", "msg", sample=0.01, sample_seed=42)

logreducer --dsn postgresql://user@host/db --query "SELECT msg FROM logs" --sample 0.01 --sample-seed 42

Notes: SQLite has no seedable RNG, so a seeded sample raises SamplingNotSupported (unseeded, best-effort sampling still works). ClickHouse SAMPLE needs the table to declare a SAMPLE BY key. For an arbitrary complex query, put the sampling clause in your own SQL instead.

Collecting a target number of lines

When you want about N representative lines rather than "reduce everything", reduce_to_target pulls fresh random batches and reduces each, accumulating distinct representatives until a stop condition fires. Peak memory is bounded to roughly one batch plus the accumulator.

flowchart TD
    PULL["pull fresh random batch<br/>(sample_batch in-engine,<br/>else client-side reservoir)"] --> E{empty?}
    E -->|yes| SE([stop: exhausted])
    E -->|no| SIZE["resize next batch from<br/>observed avg row bytes"]
    SIZE --> RED["reduce the batch"]
    RED --> ACC["add new distinct<br/>representatives"]
    ACC --> TGT{"target<br/>reached?"}
    TGT -->|yes| ST([stop: target])
    TGT -->|no| PLAT{"no new lines for<br/>plateau_rounds?"}
    PLAT -->|yes| SP([stop: plateau])
    PLAT -->|no| MEM{"memory<br/>watchdog"}
    MEM -->|"soft: shrink batch"| FET
    MEM -->|"hard limit"| SM([stop: memory])
    FET{"max_fetches<br/>hit?"} -->|yes| SF([stop: max_fetches])
    FET -->|no| PULL

from logreducer import LogReducer, reduce_to_target
from logreducer.sql import SQLSource

reducer = LogReducer(level="enhanced", mode="hybrid")
source = SQLSource("postgresql://user@host/db", "SELECT msg FROM logs")
outcome = reduce_to_target(source, reducer, target_rows=5000, max_batch_memory_gb=1.0)
print(outcome["stats"]["stop_reason"])   # target / exhausted / max_fetches / plateau / memory
lines = outcome["lines"]

logreducer --dsn postgresql://user@host/db --query "SELECT msg FROM logs" \
    --target-rows 5000 --max-fetches 50 --max-batch-memory 1.0

Bounding memory

The core streams, so pattern/temporal modes run in near-constant memory. The knobs:

• max_memory_gb - the overall cap (also sizes file read strategy and the watchdog). Presets: 0.5 / 1 / 2 GB by level; values above 70% of available RAM are clamped with a warning.
• max_clusters - LRU-bound the Drain3 template store (unbounded by default).
• anomaly_max_rows - reservoir-cap the rows fed to anomaly detection (off by default; trades anomaly recall for a bounded matrix).

reducer = LogReducer(level="enhanced", mode="hybrid", max_clusters=50_000, anomaly_max_rows=200_000)

Sources and sinks

The engine is built around two tiny structural protocols - that is the whole integration surface:

flowchart LR
    subgraph inputs ["Sources"]
        direction TB
        FS["FileSource"]
        SS["SQLSource  [sql]"]
        CS["ClickHouseSource  [clickhouse]"]
        KS["KafkaSource  [kafka]"]
        MY["your own re-iterable"]
    end
    subgraph core ["LogReducer core"]
        C["zero IO dependencies<br/>Source in -> list[str] out"]
    end
    subgraph outputs ["Sinks"]
        direction TB
        FK["FileSink"]
        KK["KafkaSink  [kafka]"]
        MS["your own write()"]
    end
    inputs -->|"__iter__() -> str<br/>fresh iterator per pass"| core
    core -->|"write(lines) -> int"| outputs

An application can hand the reducer its own IO - anything that yields str and can be iterated more than once is a Source:

class MySource:
    def __iter__(self):
        yield from open_my_stream()  # must return a FRESH iterator each call

reducer.reduce(MySource())

Output works the same way through a Sink (write(lines) -> int):

from logreducer import LogReducer, FileSink

reducer.reduce(source, sink=FileSink("reduced.jsonl", output_format="jsonl"))

Embedding in a host application

A host app that owns its own config cascade and logging standard drives logreducer with three seams - no logreducer-side knowledge of the host required:

import logreducer

# 1. Config: build it from YOUR cascade and inject it (kwargs still win on top)...
cfg = logreducer.BigDialConfig(max_memory_gb=0.5, max_patterns=800)
reducer = logreducer.LogReducer(mode="hybrid", config=cfg)

# ...or let logreducer read your env cascade directly: prefixed beats bare.
cfg = logreducer.BigDialConfig.from_env("MYAPP", "LOGREDUCER")   # MYAPP_MAX_MEMORY_GB wins

# 2. Logging: your app owns the stdlib logging handlers; logreducer attaches
#    nothing and its records flow through YOUR handlers, at YOUR level,
#    formatted by YOUR standard.
logreducer.setup_logging(enable=True, own_sinks=False)

Configuration

Override any config field as a keyword argument (unknown names raise - typos never silently no-op):

reducer = LogReducer(
    level="enhanced",
    mode="hybrid",
    max_memory_gb=1.0,          # memory ceiling (clamped to 70% of available RAM)
    dedup_cache_size=100000,    # bounded dedup cache
    drain_similarity=0.4,       # pattern similarity threshold
    fuzzy_threshold=0.8,        # fuzzy-dedup threshold (enhanced/maximum)
    anomaly_contamination=0.1,  # expected anomaly fraction
    temporal_window_minutes=60, # grouping window for temporal mode
    max_patterns=2000,          # cap on extracted patterns
)

Logging

Logging is off by default. Enable it and pick a format via env vars:

LOG_LEVEL=DEBUG LOG_FORMAT=json logreducer app.log --log

Logging is the standard library logging module, under the logreducer logger. LOG_FORMAT=json emits one JSON object per line for log aggregators (keys timestamp, level, target, function, line_number, message, fields); the default is human-readable (coloured in a terminal, plain in CI/containers). Embedding? See own_sinks=False above.

Development

git clone https://github.com/hyperi-io/logreducer.git
cd logreducer
uv sync --all-extras

See CONTRIBUTING.md for the toolchain, test layout (including the real-log corpora under tests/testdata/) and how the integration tests find their services.

License

Apache-2.0. Third-party attributions are recorded in NOTICE.

Copyright 2026 HYPERI PTY LIMITED.

Credits: template mining (src/logreducer/template_mining.py) is derived from Drain3, Copyright (c) 2020-2022 International Business Machines and the Drain3 project contributors, MIT licence. Drain3 builds on the Drain parser in LogPAI's logparser. The module header carries Drain3's licence text in full.

Context

Depth: docs/architecture.md.

What this is

A reduction engine over an abstraction, plus a CLI on top of it. The core carries zero IO dependencies and reduces any re-iterable stream of str lines, so it never owns a connection or a loading path. It is not a log shipper, not a parser, and not an aggregator front end -- it picks a representative subset of the lines it is given and drops the rest.

Where things live

Path	Holds
src/logreducer/core.py	LogReducer -- the four modes, the multi-pass orchestration, stats, output
src/logreducer/sources.py, sinks.py	the two structural protocols that ARE the integration surface, plus FileSource and FileSink
src/logreducer/sql.py, clickhouse.py, kafka.py	the optional adapters, one extra each
src/logreducer/patterns.py, anomaly.py, temporal.py	Drain3 mining and fuzzy dedup, Isolation Forest, time-window grouping
src/logreducer/template_mining.py	the Drain template miner, derived from Drain3 (credited under License)
src/logreducer/memory.py	MemoryMonitor, BoundedDeduplicator, the file read strategies
src/logreducer/sampling.py, target.py	dialect-aware SQL sampling, and the reduce_to_target batch loop
src/logreducer/config.py	BigDialConfig, the level presets, the from_env cascade
src/logreducer/cli.py	the logreducer entry point, dispatching on the --dsn scheme
tests/unit/, tests/integration/	server-free, and integration-marked against real services
tests/testdata/	gzipped, PII-cleansed slices of real public log datasets, with a manifest and a rebuild tool

Commands that prove a change

uv sync --all-extras
hyperi-ci check        # the canonical local gate -- there is no Makefile

Individually: uv run ruff format, uv run ruff check --fix, uv run mypy src/logreducer, uv run ty check src/logreducer, uv run pytest -q, uv build. Three ways green lies:

It looks like	It is actually
pytest -q covered the adapters	Nothing is deselected by default -- addopts is only -ra -q --strict-markers --strict-config, so tests/integration/ is collected and then SKIPS when there is neither a configured endpoint nor Docker. CI sets no env vars, so CI always takes the docker path and a local run may not have
the tree tells you the version	It does not. pyproject.toml and VERSION both say 3.4.0 while the newest tag is v3.5.0 and PyPI serves 3.5.0. semantic-release resolves the version at release time
coverage is at the stated floor	min_coverage: 75 gates, and its own comment says actual is 80. vulture is set to warn, not fail, because a library's public API reads as unused to dead-code analysis

What tends to bite

Don't	Do	Why
Hand reduce() a bare generator	Hand it something whose __iter__ returns a FRESH iterator each call	reduce() counts lines in one pass then re-reads to process, and hybrid reads again. A one-shot iterator is drained by the count and leaves nothing behind, so it now raises TypeError rather than returning an empty result (core.py)
Assume a reused LogReducer starts clean	Nothing -- _reset_components() rebuilds them per run	The dedup seen-set, both Drain3 miners and the fuzzy LSH accumulate per line. Carried into the next run they drop every line as already-seen and return empty (core.py)
Share one deduplicator across hybrid's two passes	Rebuild it before the anomaly pass	The pattern pass consumes the seen-set, so the anomaly pass sees everything as seen and no anomalies survive. Hybrid silently degrades to pattern-only (core.py)
Assume an unknown config keyword is ignored	Read the ValueError and fix the name	A silently dropped override means a user's tuning quietly does nothing (core.py)
Assume every template is the final Drain3 one	Allow for a cluster the max_clusters cap evicted	extract_patterns re-reads each template from the live cluster once the feed ends, so a varying cluster carries its <*> slots and the priority boost counts them. A cluster LRU-evicted mid-feed is gone by then and keeps the template it had when first seen (patterns.py)
Hand-edit the version	Land a conventional commit and let semantic-release do it	The old regime wrote VERSION by hand and took five commits to get tagging working. CONTRIBUTING makes it a standing rule

Where this sits

One declared edge in dfe-infra/suite.yaml, and it runs the surprising way.

Repo	Direction	Kind	Mechanism
dfe-engine	outbound	python-dep-undeclared	dfe-engine imports this library at runtime and declares NO dependency on it, deliberately. from logreducer import LogReducer at sampling/service.py:200, from logreducer.clickhouse import ClickHouseSource at :236, and from logreducer.kafka import KafkaSource at sampling/kafka_reader.py:134. All three are lazy, inside the gated smart and anomaly sampler modes, and an ImportError is turned into an error telling the caller to use mode=recent or mode=random instead. There is no version range to test -- install a new version alongside dfe-engine and run the tests that exercise those three sites

dfe-stack suite --consumer logreducer returns an empty edge list: no repo in that graph is something this library depends on. The runtime dependencies (psutil, numpy, scikit-learn, typer) are packages, declared in pyproject.toml, which is where they belong.

The reason dfe-engine gives for not declaring the dependency -- pin it "once it hits public PyPI" -- has been met since 3.4.0 was published on 2026-07-03, so expect that edge to become an ordinary pinned dependency.