数据工程基础

这一课你将能做到

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))