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