From JSON documents to an H5Col table — application logs#
Structured logs are semi-structured JSON: every record shares a few scalar
fields (time, level, service, message) but also carries variable parts — a list
of tags, a free-form context object, and a stack trace that appears only on
errors. Storing millions of these as one JSON blob per row makes analytics slow:
you must parse every blob to answer “how many ERRORs per service?”.
Shredding each document into columns fixes that: closed-set fields become compact categoricals, the message is a fixed-length string, and the variable parts become list columns. Then a query touches only the columns it needs, and repetitive fields (level, service) compress hard.
import json
import tempfile
from collections import Counter
from pathlib import Path
import h5py
from h5col import (
ColumnSpec,
FixedString,
ListColumnSpec,
StringValuesSpec,
Table,
bool_dtype,
)
path = Path(tempfile.gettempdir()) / "h5col_logs.h5"
Some raw log records#
Six JSON documents with the usual variety — different levels and services, empty tags/context, and stack traces only on the errors.
logs = [
{
"ts": 1_700_000_000_000,
"level": "INFO",
"service": "api",
"message": "request completed",
"handled": True,
"tags": ["http", "v1"],
"context": {"user": "u42", "route": "/things", "status": "200"},
},
{
"ts": 1_700_000_000_120,
"level": "DEBUG",
"service": "worker",
"message": "job dequeued",
"handled": True,
"tags": ["queue"],
"context": {"job_id": "9f3", "attempt": "1"},
},
{
"ts": 1_700_000_000_450,
"level": "WARNING",
"service": "api",
"message": "slow query",
"handled": True,
"tags": ["http", "db", "slow"],
"context": {"route": "/search", "latency_ms": "812"},
},
{
"ts": 1_700_000_000_900,
"level": "ERROR",
"service": "worker",
"message": "unhandled exception while processing job",
"handled": False,
"tags": ["job", "exception"],
"context": {"job_id": "9f3", "attempt": "3"},
"stack_trace": [
"Traceback (most recent call last):",
" File 'worker.py', line 88, in run",
" File 'tasks.py', line 41, in process",
"ValueError: bad payload",
],
},
{
"ts": 1_700_000_001_050,
"level": "INFO",
"service": "api",
"message": "request completed",
"handled": True,
"tags": [],
"context": {},
},
{
"ts": 1_700_000_001_500,
"level": "CRITICAL",
"service": "db",
"message": "replica connection lost",
"handled": False,
"tags": ["db", "replication", "alert"],
"context": {"replica": "db-2", "lag_s": "37"},
"stack_trace": ["ConnectionResetError: peer reset"],
},
]
print(json.dumps(logs[3], indent=2))
{
"ts": 1700000000900,
"level": "ERROR",
"service": "worker",
"message": "unhandled exception while processing job",
"handled": false,
"tags": [
"job",
"exception"
],
"context": {
"job_id": "9f3",
"attempt": "3"
},
"stack_trace": [
"Traceback (most recent call last):",
" File 'worker.py', line 88, in run",
" File 'tasks.py', line 41, in process",
"ValueError: bad payload"
]
}
The mapping#
JSON field |
Shape |
H5Col column |
|---|---|---|
|
scalar int |
|
|
small ordered set |
ordered categorical |
|
small set |
categorical |
|
short text |
|
|
boolean |
|
|
array of strings |
|
|
variable key→value object |
two aligned |
|
array of strings, only on errors |
nullable |
A JSON object with arbitrary keys has no fixed column per key, so we keep it
lossless and columnar as two aligned lists — context_keys[i] and
context_values[i] zip back into the original dict. (You’d typically promote
well-known keys to their own columns and leave the long tail in the maps.)
# Shred the documents into per-column arrays.
ts = [r["ts"] for r in logs]
level = [r["level"] for r in logs]
service = [r["service"] for r in logs]
message = [r["message"] for r in logs]
handled = [r["handled"] for r in logs]
tags = [r.get("tags", []) for r in logs]
ctx_keys = [list(r.get("context", {}).keys()) for r in logs]
ctx_vals = [list(map(str, r.get("context", {}).values())) for r in logs]
stack = [r.get("stack_trace") for r in logs] # None where absent -> null list
f = h5py.File(path, "w")
t = Table.create(
f.create_group("logs"),
[
ColumnSpec(
name="ts",
dtype="i8",
units="ms",
description="event time, epoch milliseconds",
),
ColumnSpec(
name="level",
ordered=True,
categories=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
),
ColumnSpec(name="service", categories=["api", "worker", "db"]),
ColumnSpec(name="message", dtype=FixedString(120)),
ColumnSpec(name="handled", dtype=bool_dtype()),
ListColumnSpec(name="tags", values=StringValuesSpec()),
ListColumnSpec(name="context_keys", values=StringValuesSpec()),
ListColumnSpec(name="context_values", values=StringValuesSpec()),
ListColumnSpec(
name="stack_trace",
values=StringValuesSpec(),
nullable=True,
description="frames; null unless an error occurred",
),
],
title="Application log records",
index_columns=["ts"],
)
t.append(
{
"ts": ts,
"level": level,
"service": service,
"message": message,
"handled": handled,
"tags": tags,
"context_keys": ctx_keys,
"context_values": ctx_vals,
"stack_trace": stack,
}
)
print("stored", t.nrows, "records ·", len(t.column_names), "columns")
stored 6 records · 9 columns
Round-trip: rebuild a JSON document from the columns#
The columnar form is lossless — we can reassemble any record, including its
context object and (only where present) its stack_trace.
def reconstruct(table, i):
d = table.read()
rec = {
"ts": int(d["ts"][i]),
"level": d["level"][i],
"service": d["service"][i],
"message": d["message"][i],
"handled": bool(d["handled"][i]),
"tags": list(d["tags"][i]),
"context": dict(zip(d["context_keys"][i], d["context_values"][i], strict=True)),
}
frames = d["stack_trace"][i]
if frames is not None: # null list -> field simply absent
rec["stack_trace"] = list(frames)
return rec
print(json.dumps(reconstruct(t, 3), indent=2)) # the ERROR record
{
"ts": 1700000000900,
"level": "ERROR",
"service": "worker",
"message": "unhandled exception while processing job",
"handled": false,
"tags": [
"job",
"exception"
],
"context": {
"job_id": "9f3",
"attempt": "3"
},
"stack_trace": [
"Traceback (most recent call last):",
" File 'worker.py', line 88, in run",
" File 'tasks.py', line 41, in process",
"ValueError: bad payload"
]
}
Analytics the columnar form makes cheap#
No JSON parsing: counts read one small categorical column, and the ordered
level lets us select “ERROR and above” by code.
print("records by level:", dict(Counter(t["level"].read())))
lvl = t["level"]
severity = list(lvl.categories) # ordered: DEBUG < ... < CRITICAL
min_code = severity.index("ERROR")
codes = lvl.codes
msgs = t["message"].read()
svc = t["service"].read()
print("\nERROR and above:")
for i in range(t.nrows):
if int(codes[i]) >= min_code:
print(f" {severity[int(codes[i])]:8s} {svc[i]:6s} {msgs[i]}")
records by level: {'INFO': 2, 'DEBUG': 1, 'WARNING': 1, 'ERROR': 1, 'CRITICAL': 1}
ERROR and above:
ERROR worker unhandled exception while processing job
CRITICAL db replica connection lost
Storage note#
Chunk shapes come from h5col’s cache-aware default, so they’re sized for scale
(millions of rows) even though this demo holds a handful — HDF5 allocates chunk
storage lazily as rows are written.
print("message chunk length:", t["message"].dataset.chunks[0])
print("tags/CHARS chunk length:", t["tags"].group["VALUES/CHARS"].chunks[0])
t.validate()
print("valid H5Col table ✔")
f.close()
message chunk length: 34952
tags/CHARS chunk length: 4194304
valid H5Col table ✔