Log Analysis and Diagnostic Data Gathering Questions

Direct answer

Read the file one line at a time instead of loading it whole, parse each line as a JSON record, skip lines where error_type is null (those are successful inferences, not errors), and accumulate counts into a small dictionary keyed by (hour, error_type). Memory stays proportional to the number of distinct hour/error-type combinations, not to file size, since each line is discarded immediately after updating the counters.

Structured elaboration

• Streaming: iterating a file object with for line in f: reads and yields one line at a time under the hood; at no point does the script hold the whole file in memory, only the current line and the running counters.
• Memory bound: the aggregate structure is a dict of Counters, one entry per hour bucket, each holding one count per distinct error_type seen in that hour. Its size is bounded by (number of hours) x (number of distinct error types), which stays small even for a file with billions of lines, since neither of those dimensions grows with file size.
• Assumptions: timestamps are ISO8601 (YYYY-MM-DDTHH:MM:SSZ or with a numeric offset) and treated as UTC once parsed; a null (or missing) error_type means the inference succeeded and is intentionally excluded from the error counts (only present to establish "no error," not counted as its own category); malformed lines (bad JSON, missing timestamp) are skipped and counted separately rather than crashing the whole run, so one corrupt line doesn't lose the rest of a multi-gigabyte file's results.

Worked example

import gzip, json
from collections import defaultdict, Counter
from datetime import datetime

def open_stream(path):
    return gzip.open(path, "rt", encoding="utf-8") if path.endswith(".gz") \
        else open(path, "r", encoding="utf-8")

def hour_bucket(ts):
    dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
    return dt.strftime("%Y-%m-%dT%H:00")

def count_errors(path):
    counts = defaultdict(Counter)   # {hour: Counter({error_type: count})}
    skipped = 0
    with open_stream(path) as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                rec = json.loads(line)
            except json.JSONDecodeError:
                skipped += 1
                continue
            ts = rec.get("timestamp")
            err = rec.get("error_type")
            if not ts:
                skipped += 1
                continue
            if err is None:
                continue   # successful inference, not an error
            counts[hour_bucket(ts)][err] += 1
    return counts, skipped

I ran this against a pinned, gzip-compressed synthetic file: 6 well-formed records spanning two hourly buckets (4 with an error_type, 2 successful with error_type: null), plus 1 deliberately malformed line:

records = [
    {"timestamp": "2026-01-01T09:05:00Z", "model": "fraud-v3", "error_type": None, "latency_ms": 42},
    {"timestamp": "2026-01-01T09:12:00Z", "model": "fraud-v3", "error_type": "timeout", "latency_ms": 5000},
    {"timestamp": "2026-01-01T09:40:00Z", "model": "fraud-v3", "error_type": "timeout", "latency_ms": 5000},
    {"timestamp": "2026-01-01T09:50:00Z", "model": "fraud-v3", "error_type": "schema_mismatch", "latency_ms": 12},
    {"timestamp": "2026-01-01T10:02:00Z", "model": "fraud-v3", "error_type": None, "latency_ms": 38},
    {"timestamp": "2026-01-01T10:15:00Z", "model": "fraud-v3", "error_type": "timeout", "latency_ms": 5000},
]
with gzip.open("inference.ndjson.gz", "wt", encoding="utf-8") as f:
    for r in records:
        f.write(json.dumps(r) + "\n")
    f.write("not even json\n")   # 1 deliberately malformed line

counts, skipped = count_errors("inference.ndjson.gz")
print("hour,error_type,count")
for hour in sorted(counts):
    for err, c in counts[hour].most_common():
        print(f"{hour},{err},{c}")
print(f"skipped malformed/incomplete lines: {skipped}")

Output:

hour,error_type,count
2026-01-01T09:00,timeout,2
2026-01-01T09:00,schema_mismatch,1
2026-01-01T10:00,timeout,1
skipped malformed/incomplete lines: 1

That matches the input by hand: 3 error records land in the 09:00 bucket (2 timeout, 1 schema_mismatch), 1 error record lands in the 10:00 bucket (timeout), the 2 successful (null error_type) records are correctly excluded from the error counts, and the 1 malformed line is correctly counted as skipped rather than crashing the run or silently vanishing.

Running this on large files: streaming, gz, S3

• Streaming is already the default here: for line in f never materializes the whole file, so a 500 GB file runs with the same memory footprint as a 500 MB one.
• gz support is a one-line branch (gzip.open(..., "rt") instead of open(...)), shown above; gzip's file object is itself iterable line-by-line, so it doesn't require decompressing to a temp file first.
• S3 input, for a file too large or inconvenient to download first, means streaming directly from the object store instead of a local path: boto3's get_object() returns a StreamingBody, which can be wrapped in io.TextIOWrapper (and, if the object is also gzip-compressed, layered through gzip.GzipFile(fileobj=...)) so the same line-by-line loop above works unchanged, without ever writing the object to local disk.

Trade-offs & pitfalls

• Treating null error_type as "no error" is an assumption stated explicitly, not verified from data; if the real pipeline sometimes uses an empty string or a sentinel value like "none" instead of a genuine null, this logic silently miscounts those as a distinct error type. Worth confirming against real sample data before trusting this at scale.
• Bucketing purely by hour with no timezone normalization check is fine if every producer genuinely writes UTC; a single service accidentally logging local time would corrupt the hourly buckets in a way this script has no way to detect on its own.
• For extremely high cardinality of error_type values (say, thousands of distinct free-text error strings) the in-memory counters could grow large enough to matter; if that's a real risk, aggregating incrementally to an external store (a database, or spilling to disk) per hour as you go, rather than holding every hour's counters for the whole run, keeps the memory bound tight regardless of cardinality.

Direct answer

Before writing this query, there's a clarifying question worth asking out loud: the stated schema for inference_logs has no predicted-label or score column, and without one, "accuracy" (was the prediction right) genuinely cannot be computed, only "label coverage" (did a true label arrive at all) can. I'll state that as an explicit assumption and proceed as if inference_logs also has a predicted_label column, which is the realistic case this question is almost certainly modeling. GROUP BY model_version and an hourly bucket of the timestamp, LEFT JOIN to labels on request_id so unlabeled requests are still counted, and compute accuracy only over the subset that has a matching label.

Structured elaboration

• Join choice: LEFT JOIN inference_logs -> labels on request_id, not an inner join. Labels typically arrive after the prediction (someone has to observe the real outcome), so at query time some recent requests won't have a label yet. An inner join would silently drop those rows from the count and mean latency too, which is wrong: latency and volume are properties of every request, whether or not a label has shown up.
• Handling missing labels: a request with no matching labels row gets true_label IS NULL after the left join. It still counts toward requests_count and contributes to mean latency_ms, but it's excluded from both the numerator and denominator of the accuracy calculation. If a model_version/hour bucket has zero labeled requests, accuracy should return NULL (unknown), not 0, since 0 would falsely read as "this model is always wrong" rather than "no ground truth exists yet."
• Assumptions about timestamps: the query assumes both tables' timestamps are directly comparable (same timezone, effectively UTC) and buckets by inference_logs.timestamp (when the prediction was made), not the label's arrival time, since the question asks for a per-hour view of model behavior, which should be indexed by when the model acted, not by whenever ground truth happened to show up. The join matches purely on request_id, deliberately not also requiring the label's timestamp to fall in the same hour, since labels routinely arrive after a delay (label lag) and requiring same-hour timestamps would exclude real, valid labels for no good reason.

Worked example

SELECT
  date_trunc('hour', il.timestamp) AS hour_bucket,
  il.model_version,
  COUNT(*) AS requests_count,
  AVG(il.latency_ms) AS mean_latency_ms,
  -- FILTER (WHERE ...) is standard Postgres aggregate-filter syntax: it scopes
  -- an aggregate to only the rows matching the condition, without a subquery.
  CASE WHEN COUNT(*) FILTER (WHERE l.true_label IS NOT NULL) = 0 THEN NULL
       ELSE COUNT(*) FILTER (WHERE l.true_label IS NOT NULL
                               AND il.predicted_label = l.true_label)::numeric
            / COUNT(*) FILTER (WHERE l.true_label IS NOT NULL)
  END AS accuracy
FROM inference_logs il
LEFT JOIN labels l ON il.request_id = l.request_id
GROUP BY hour_bucket, il.model_version
ORDER BY hour_bucket, il.model_version;

I don't have a live Postgres instance in this environment to run the exact syntax above against, so I verified the join, grouping, and null-handling logic with an equivalent query in SQLite (which lacks date_trunc/FILTER but has identical join semantics) against pinned synthetic data: 6 requests across 2 hourly buckets and 2 model versions, with 2 of the 6 requests deliberately missing a label (simulating label lag):

import sqlite3
conn = sqlite3.connect(":memory:")
cur = conn.cursor()
cur.execute("CREATE TABLE inference_logs (ts TEXT, request_id TEXT, model_version TEXT, predicted_label TEXT, latency_ms INTEGER)")
cur.execute("CREATE TABLE labels (ts TEXT, request_id TEXT, true_label TEXT)")
cur.executemany("INSERT INTO inference_logs VALUES (?,?,?,?,?)", [
    ("2026-01-01T10:05:00", "r1", "v1", "spam", 120),
    ("2026-01-01T10:20:00", "r2", "v1", "ham",  95),
    ("2026-01-01T10:40:00", "r3", "v2", "spam", 200),
    ("2026-01-01T11:02:00", "r4", "v1", "spam", 130),
    ("2026-01-01T11:15:00", "r5", "v2", "ham",  180),
    ("2026-01-01T11:50:00", "r6", "v2", "ham",  210),
])
cur.executemany("INSERT INTO labels VALUES (?,?,?)", [
    ("2026-01-01T10:06:00", "r1", "spam"),  # correct
    ("2026-01-01T10:21:00", "r2", "spam"),  # wrong (predicted ham, true spam)
    ("2026-01-01T11:03:00", "r4", "spam"),  # correct
    ("2026-01-01T11:16:00", "r5", "ham"),   # correct
    # r3, r6: no label yet (label lag)
])
query = (
    "SELECT strftime('%Y-%m-%d %H:00', il.ts) AS hour_bucket, il.model_version, "
    "COUNT(*) AS n, AVG(il.latency_ms) AS mean_lat, "
    "CASE WHEN SUM(CASE WHEN l.true_label IS NOT NULL THEN 1 ELSE 0 END) = 0 THEN NULL "
    "ELSE CAST(SUM(CASE WHEN l.true_label IS NOT NULL AND il.predicted_label = l.true_label "
    "THEN 1 ELSE 0 END) AS REAL) "
    "/ SUM(CASE WHEN l.true_label IS NOT NULL THEN 1 ELSE 0 END) END AS accuracy "
    "FROM inference_logs il LEFT JOIN labels l ON il.request_id = l.request_id "
    "GROUP BY hour_bucket, il.model_version ORDER BY hour_bucket, il.model_version"
)
print(f"{'hour':<18} {'model':<7} {'count':>6} {'mean_latency':>13} {'accuracy':>9}")
for hour, model, n, mean_lat, acc in cur.execute(query):
    print(f"{hour:<18} {model:<7} {n:>6} {mean_lat:>13.1f} {(f'{acc:.3f}' if acc is not None else 'NULL'):>9}")

Output:

hour               model    count  mean_latency  accuracy
2026-01-01 10:00   v1           2         107.5     0.500
2026-01-01 10:00   v2           1         200.0      NULL
2026-01-01 11:00   v1           1         130.0     1.000
2026-01-01 11:00   v2           2         195.0     1.000

Checking by hand: the 10:00, v1 bucket has 2 requests (correctly counted), one predicted correctly and one wrong, giving accuracy 0.5, exactly as computed. The 10:00, v2 bucket has 1 request with no label at all, so it's still counted (mean latency = 200.0) but accuracy correctly returns NULL rather than 0 or 1, since there's no ground truth to score it against yet.

Trade-offs & pitfalls

• If predicted_label genuinely doesn't exist on inference_logs in the real schema, this query has to be reframed as label-coverage reporting (what fraction of requests eventually got a label, and how long that took) rather than accuracy; that's a materially different question, and confirming which one is actually needed is worth the clarifying question before writing any SQL.
• Multiple label rows for the same request_id (a correction, or a re-labeling pass) would silently multiply that request's contribution to requests_count after the join; deduplicating labels to the latest one per request_id before joining (a DISTINCT ON (request_id) ... ORDER BY timestamp DESC subquery in Postgres) avoids that.
• Bucketing by the prediction's own timestamp means a model's accuracy for "this hour" will keep changing retroactively as more labels arrive; that's expected behavior for label lag, but it means this query's output for the most recent few hours should be read as provisional, not final.