Processing Log Data
Treat a log-processing script as a small data pipeline:
bytes → decoded records → parsed fields → validated events → aggregation → output
Define what happens to malformed records at each boundary. Silently dropping them makes operational conclusions difficult to trust.
Stream and parse
Iterate over the input rather than loading an unbounded log into memory. Prefer a documented structured format such as JSON Lines. For a fixed legacy text format, use a compiled pattern with named fields. Here is the JSON Lines version:
import json
from collections import Counter
from pathlib import Path
counts: Counter[str] = Counter()
bad_records = 0
with Path("events.jsonl").open("r", encoding="utf-8") as handle:
for line_number, line in enumerate(handle, start=1):
try:
event = json.loads(line)
user = event["user"]
if not isinstance(user, str) or not user:
raise ValueError("invalid user")
except (json.JSONDecodeError, KeyError, ValueError, TypeError):
bad_records += 1
continue
counts[user] += 1
The policy may fail fast, quarantine bad records, or continue with a counter. Choose it based on whether incomplete results remain meaningful, and expose the number of rejected records in the output or metrics.
Operational boundaries
- Logs may rotate, truncate, arrive out of order, repeat, or end with a partially written record.
- Timestamps need an explicit format and timezone. Parse into aware datetime values when comparisons cross systems.
- Usernames, paths, and messages may contain sensitive information. Minimize, redact, and control retention at ingestion rather than after copying them.
- Regex parsing encodes a schema even if no schema file exists. Version and test that contract when producers can change.
- Exact aggregation keeps one key per distinct value. High-cardinality or unbounded keys require limits, external storage, or approximate algorithms.
For a one-shot batch, emit deterministic output such as sorted JSON or CSV. For continuous processing, add checkpoints or an idempotency strategy so restarting does not double count. A filesystem offset alone may be invalid after rotation.
Producing logs
Application modules should obtain logging.getLogger(__name__); the application
entry point owns handler and format configuration. Machine-consumed logs should
carry stable fields rather than forcing later tools to recover structure from a
human sentence.
Logs are evidence, not ground truth. Record parser version, input scope, rejected record count, and processing time when the result informs an operational or security decision.
A bounded batch example
The code above parses JSON Lines, not the legacy regex format. Save these four lines as events.jsonl in a scratch directory and run the example there:
{"user": "ada"}
{"user": "ada"}
{"user": "grace"}
{"user": ""}
Add this after the loop to expose both results and rejected records:
print(json.dumps({"counts": dict(counts), "bad_records": bad_records}, sort_keys=True))
The result is {"bad_records": 1, "counts": {"ada": 2, "grace": 1}}. Valid JSON alone does not guarantee an event object: null, [], a missing user, and a numeric user are rejected too. json.loads raises JSONDecodeError, a subclass of ValueError, for invalid JSON.
This example counts parsing/schema failures but fails the batch on file access or UTF-8 decoding errors. Decoding happens while advancing the file iterator, outside the inner try. Streaming bounds retained input to roughly a line, not a fixed byte limit: a huge single record still needs a size policy. The Counter also grows with distinct users.