日志数据处理
将日志处理脚本视为一条微型数据管道：
字节流 → 解码记录 → 解析字段 → 校验事件 → 聚合 → 输出
明确每个边界如何处理畸形记录。静默丢弃坏数据会让后续的运维结论失去可信度。
流式读取与解析
遍历输入流，切勿将无界日志一次性加载进内存。优先采用 JSON Lines 等有明确文档的结构化格式。若必须处理固定的遗留文本格式，可使用带命名捕获组的编译正则表达式。下面演示 JSON Lines 版本：
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
处理策略可以是快速失败（Fail-fast）、隔离坏记录，或计数后继续。选择哪种策略取决于不完整的结果是否仍有业务价值，并务必在输出或监控指标中暴露被拒绝的记录数。
运维边界考量
- 日志特性：日志可能轮转、截断、乱序到达、重复，甚至以未写完的记录结尾。
- 时间戳：需明确格式与时区。跨系统比较时，务必解析为带时区信息的
datetime对象。 - 敏感信息：用户名、路径和消息体可能包含敏感数据。应在摄入阶段进行最小化、脱敏和保留期控制，而非在数据复制后再处理。
- 隐式 Schema：正则解析本质上是在编码一种 Schema。即使没有显式的 Schema 文件，当上游生产者可能变更时，也需对该契约进行版本管理和测试。
- 聚合策略：精确聚合会为每个唯一值保留一个键。面对高基数或无界键时，需引入限制、外部存储或近似算法。
对于一次性批处理，输出确定性结果（如排序后的 JSON 或 CSV）。对于持续处理，需引入检查点或幂等策略，防止重启导致重复计数。注意：仅依赖文件系统偏移量在日志轮转后可能失效。
日志生成规范
应用模块应通过 logging.getLogger(__name__) 获取 Logger；应用入口点负责 Handler 和格式配置。供机器消费的日志应携带稳定的结构化字段，避免迫使下游工具从人类可读的自然语言中逆向解析结构。
日志是证据，而非绝对真理。当结果用于运维或安全决策时，应记录解析器版本、输入范围、拒绝记录数及处理耗时。
可运行的批处理示例
上面的代码解析 JSON Lines，而不是前文提到的旧式正则格式。在临时工作目录中把下面四行保存为 events.jsonl，再运行示例：
{"user": "ada"}
{"user": "ada"}
{"user": "grace"}
{"user": ""}
在循环结束后添加输出，同时展示计数和拒绝记录数：
print(json.dumps({"counts": dict(counts), "bad_records": bad_records}, sort_keys=True))
结果为 {"bad_records": 1, "counts": {"ada": 2, "grace": 1}}。合法 JSON 不一定是合法事件：null、[]、缺少 user 或 user 为数字的记录也会被拒绝。json.loads 遇到非法 JSON 会抛出 JSONDecodeError，它是 ValueError 的子类。
此例对解析和字段校验失败计数，但文件访问失败或 UTF-8 解码错误会使批处理失败。解码发生在推进文件迭代器时，位于内部 try 之外。流式读取让保留的输入大致以一行为单位，并没有固定字节上限：单条超大记录仍需大小限制。Counter 也会随不同用户数增长。