数据工程基础

这一课你将能做到

1. 区分数据湖、数据仓库、湖仓一体，按负载选择存储与查询引擎
2. 用 pandas 写一条「原始日志 → 清洗校验 → 特征表」的可复现管道，产出 Parquet 文件
3. 启动单机 Kafka，用生产者/消费者跑通流式数据的最小闭环

为什么要学

模型准确率的上限由数据决定，代码只是逼近这个上限。MLOps 实践里大部分线上故障不在模型，而在数据：日志丢字段、上游改了 schema、统计口径悄悄变了。所以在动手训练之前，你得先有一条可靠的数据管道。

人话：判题机每判一次提交就写一条日志。这些日志是「生肉」——有脏数据、字段会变。数据工程就是把生肉加工成「净菜」（干净明细表）和「配好的菜」（特征表），模型只负责炒菜。本课为判题超时预测模型和题目难度评分模型准备它们共同的特征底座：题目维度的每日判题统计。

先建立图像

	数据湖 Lake	数据仓库 Warehouse
存什么	原始数据原样落盘（JSON/日志/Parquet）	建模后的结构化表（维度/事实表）
schema	读时定义（存进去什么样就是什么样）	写时定义（入表前必须符合表结构）
谁用	数据科学探索、历史重放	BI 报表、特征生产
典型载体	S3/OSS + Parquet 目录	MySQL / ClickHouse / StarRocks

判题机日志(jsonl) ──批：每日跑批──▶ ODS 原始层 ──清洗──▶ DWD 明细层 ──聚合──▶ ADS 特征层 ──▶ 训练/打分
        └──流：Kafka 实时──▶ Flink/Spark Streaming ──▶ 实时特征（如近 1 小时 TLE 率）

跟着做

步骤 1：准备环境

pip install pandas pyarrow confluent-kafka

步骤 2：模拟一批判题日志

# src/make_dataset.py
import json, random
from datetime import datetime, timedelta
from pathlib import Path

random.seed(42)
Path("data/raw").mkdir(parents=True, exist_ok=True)

with open("data/raw/judge_logs.jsonl", "w", encoding="utf-8") as f:
    base = datetime(2026, 1, 1)
    for i in range(5000):
        rec = {
            "submit_id": i,
            "submit_time": (base + timedelta(minutes=i * 3)).isoformat(),
            "user_id": random.randint(1, 800),
            "problem_id": random.randint(1000, 1020),
            "lang": random.choice(["Java", "C++", "Python"]),
            "code_lines": random.randint(5, 300),
            "rank": random.randint(1, 5),
            "verdict": random.choices(
                ["AC", "TLE", "WA", "RE"], weights=[55, 15, 25, 5])[0],
            "time_ms": random.randint(50, 6000),
        }
        f.write(json.dumps(rec, ensure_ascii=False) + "\n")
print("done")

python src/make_dataset.py

步骤 3：清洗并校验 schema

# src/clean.py
from pathlib import Path
import pandas as pd

ALLOWED = {
    "submit_id": "int64", "submit_time": "object",
    "problem_id": "int64", "lang": "object",
    "code_lines": "int64", "rank": "int64",
    "verdict": "object", "time_ms": "int64",
}

def load_and_validate(path: str) -> pd.DataFrame:
    df = pd.read_json(path, lines=True)
    missing = set(ALLOWED) - set(df.columns)
    if missing:
        raise ValueError(f"缺字段: {missing}")   # schema 校验：脏数据在此拦截
    df = df[list(ALLOWED)].astype(ALLOWED)
    before = len(df)
    df = df[(df["code_lines"] > 0) & (df["time_ms"] <= 10_000)]
    if df.empty:
        raise ValueError("清洗后为空，检查上游")
    print(f"剔除异常行 {before - len(df)} / {before}")
    return df

if __name__ == "__main__":
    df = load_and_validate("data/raw/judge_logs.jsonl")
    Path("data/processed").mkdir(parents=True, exist_ok=True)
    df.to_parquet("data/processed/judge_clean.parquet", index=False)
    print(df["verdict"].value_counts(normalize=True).round(3))

步骤 4：聚合出题目维度的训练特征

# src/features.py
import pandas as pd

df = pd.read_parquet("data/processed/judge_clean.parquet")
df["is_tle"] = (df["verdict"] == "TLE").astype(int)
df["day"] = pd.to_datetime(df["submit_time"]).dt.normalize()

daily = (df.groupby(["problem_id", "day"])
           .agg(submits=("submit_id", "count"),
                tle_rate=("is_tle", "mean"),
                avg_time=("time_ms", "mean"))
           .reset_index()
           .sort_values(["problem_id", "day"]))

# 近 7 天滚动 TLE 率：题目难度评分模型的核心特征之一
daily["tle_rate_7d"] = (daily.groupby("problem_id")["tle_rate"]
                          .transform(lambda s: s.rolling(7, min_periods=1).mean()))

daily.to_parquet("data/processed/problem_daily.parquet", index=False)
print(daily.tail())

这张 problem_daily.parquet 就是后续课程训练两个模型的数据源：难度评分模型用 tle_rate_7d、avg_time 做回归特征；超时预测模型用它补充题目侧统计特征。

步骤 5：体验流式链路（Kafka 单机）

docker run -d --name kafka -p 9092:9092 apache/kafka:3.9.0

# src/stream_produce.py —— 把日志文件当消息流发出去
from confluent_kafka import Producer

p = Producer({"bootstrap.servers": "localhost:9092"})
with open("data/raw/judge_logs.jsonl", encoding="utf-8") as f:
    for line in f:
        p.produce("judge-events", value=line.encode())
p.flush()
print("sent")

# src/stream_consume.py
import json
from confluent_kafka import Consumer

c = Consumer({"bootstrap.servers": "localhost:9092",
              "group.id": "feature-job", "auto.offset.reset": "earliest"})
c.subscribe(["judge-events"])
try:
    while True:
        msg = c.poll(1.0)
        if msg and not msg.error():
            rec = json.loads(msg.value())
            if rec["verdict"] == "TLE":
                print("TLE 实时事件:", rec["problem_id"])
finally:
    c.close()

两个终端分别跑 consume、produce，你能亲眼看到「日志 → 消息 → 实时处理」。生产环境中，消费者位置通常是 Flink/Spark Streaming 做窗口聚合，产出低延迟实时特征。

讲清楚要点

1. 分层是管道的骨架：ODS 原样留存（可重放）、DWD 清洗明细、ADS 面向用途聚合。出问题时每层可单独排查、单独重算。
2. schema 校验要前置：在管道入口拒绝脏数据（缺字段、类型不对、数值越界），比训练时才发现便宜一个数量级。
3. Parquet 优先于 CSV：列存、带类型、压缩、可分片。CSV 每次读取都要重新猜类型，是特征口径漂移的常见根源。
4. 批流分工与口径一致：批处理管准确的全量重算，流处理管低延迟增量特征；两边必须用同一份特征定义，否则出现 train/serve skew。
5. 特征表要可复现：同样的输入窗口 + 同样的代码 → 同样的输出。窗口长度等参数写成配置（第 3 课进 params.yaml），不要散落在代码里。

你可能会懵的地方

• 特征商店（Feature Store）现在就要建吗？ 不要。先用「特征表 + 共享代码库」解决大部分问题。当你发现训练和线上各自实现同一特征、并因此出过 bug，再引入 Feast 等工具不迟。
• Kafka 和 RabbitMQ 的区别？为什么 ML 场景常提 Kafka？ Kafka 是分区日志：消息可重放、多个消费者组独立消费，适合「同一条判题日志喂给实时特征、监控、归档多个下游」。RabbitMQ 是队列语义，消费即走，不适合重放。
• 为什么不用 MySQL 直接给模型出特征？ 简单聚合可以，但滚窗、多表 join、大规模重算会让 SQL 越写越痛。湖存原始 + Parquet 中间格式是更通用的组合；聚合查询特别重时上 ClickHouse。
• 数据漂移是这课管还是监控课管？ 监控课负责发现漂移；这课负责让数据可追溯、可重放——没有分层留存，漂移发生时你连历史数据都拿不出来对比。
• read_json 对大文件扛不住怎么办？ 分块读取逐块落 Parquet，或换 Polars/DuckDB。思路一致：入口分块、中间列存、聚合下推。

错法对照

错法	问题	正解
在 notebook 里从原始日志一路做到训练	不可复现、无法排错	拆成 make/clean/features 独立脚本
特征用 CSV 存，读回来数值列变 object	类型静默错乱	用 Parquet 并显式声明 schema
清洗时顺手丢弃异常行且不记录	数据量悄悄缩水无人知晓	记录剔除数量与原因，超阈值告警
训练特征与线上特征两套代码各写一遍	train/serve skew，线上效果劣化	同一份特征代码，两边调用
滚窗用「当前系统时间」往前推	重跑结果不同	用数据自带的业务时间字段切窗口

卡点急救

1. to_parquet 报缺少引擎：pip install pyarrow，pandas 读写 Parquet 依赖它。
2. Kafka 连不上：docker ps 确认容器在跑、9092 端口未被占用；代码里 bootstrap.servers 与映射端口一致。
3. rolling(7) 结果异常：groupby 后先 sort_values(["problem_id", "day"]) 再滚动；min_periods=1 保证窗口不足 7 天时也有值。

面试常问

• 数据湖和数据仓库怎么选？ 探索性、schema 多变、半结构化数据 → 湖；强一致性聚合、BI 报表 → 仓。主流做法是湖仓一体：湖上用开放表格式（Parquet/Iceberg）存数据，用 Spark/Trino 按需建模查询。
• 怎么保证训练特征和线上特征一致？ 单一实现：特征逻辑抽成共享包，离线跑历史窗口、在线跑实时窗口用同一份代码；两边同一套 schema 校验；再加离线回放比对——对同一批历史请求，离线与在线特征输出必须逐值一致。
• 描述一条你做过的数据管道。 按「源 → ODS → 清洗 → 特征 → 消费方」的顺序讲，重点强调三件事：入口 schema 校验、失败可重跑、按分区幂等覆盖写。

自测三问

1. ODS/DWD/ADS 各存什么、分别给谁用？
2. 特征中间格式为什么选 Parquet 而不是 CSV？
3. 什么是 train/serve skew？给出至少两条工程预防手段。

小结一句话

数据工程的职责是把原始日志变成带 schema 校验、可复现、批流口径一致的特征表，模型只是这条管道的下游消费者。

延伸

• Apache Spark 官方文档——数据量超过单机 pandas 后，批处理的标准答案。
• Apache Kafka 官方文档——分区、消费组、重放机制的权威说明。
• 下一课：版本控制与 DVC——数据集本身怎么像代码一样打版本、可回滚。