教育数据仓库的设计：从学习行为日志到多维分析

教育数据仓库的设计：从学习行为日志到多维分析

一、深度引言与场景痛点：学生在平台上做了什么，你真的"知道"吗？

在线教育平台每天都在产生大量的用户行为数据：点击了哪个课程、看了多久视频、在哪道题上卡住了、提交了什么答案。但大多数平台只是把这些数据存起来，用于"用户看了 3 个视频"这类基础统计。

真正的价值在于多维分析——将行为数据按时间、知识域、用户群体等维度交叉分析。例如："上周数学成绩下降最明显的是上午 10 点上课的学生群体"这种洞察，需要将用户行为日志、题目表现数据和课程信息三张表关联分析。

数据仓库（而非关系数据库）正是为这种多维分析场景设计的。

二、底层机制与原理深度剖析

星型模型设计

三、生产级代码实现与最佳实践

# 教育数据仓库 ETL 管道
from datetime import datetime, timedelta
class EducationDataWarehouse:
    """教育数据仓库 —— ETL 和查询层
    采用星型模型，以"学习行为"作为事实表，
    学生、题目、课程、时间作为维度表。
    """
    def __init__(self, db_connection):
        self.db = db_connection
    def build_dim_tables(self):
        """构建维度表"""
        # 时间维度 —— 预生成日期数据，方便按年/季度/月/周查询
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS dim_time (
                time_id INT PRIMARY KEY,
                full_date DATE,
                year INT,
                quarter INT,
                month INT,
                week_of_year INT,
                day_of_week INT,
                hour INT,
                is_weekend BOOLEAN
            )
        """)
        # 学生维度
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS dim_student (
                student_id VARCHAR(32) PRIMARY KEY,
                grade VARCHAR(20),
                city VARCHAR(50),
                school VARCHAR(100),
                registered_date DATE,
                user_type VARCHAR(20) COMMENT '付费/免费/试用'
            )
        """)
        # 题目维度
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS dim_problem (
                problem_id VARCHAR(32) PRIMARY KEY,
                problem_type VARCHAR(20) COMMENT '选择题/填空题/解答题',
                difficulty VARCHAR(10) COMMENT 'EASY/MEDIUM/HARD',
                subject VARCHAR(20),
                knowledge_point VARCHAR(50),
                chapter VARCHAR(50),
                score INT
            )
        """)
        # 课程维度
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS dim_course (
                course_id VARCHAR(32) PRIMARY KEY,
                course_name VARCHAR(100),
                subject VARCHAR(20),
                grade_level VARCHAR(20),
                teacher_name VARCHAR(50),
                total_lessons INT,
                created_date DATE
            )
        """)
    def build_fact_table(self):
        """构建事实表"""
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS fact_learning_behavior (
                behavior_id BIGINT AUTO_INCREMENT PRIMARY KEY,
                student_id VARCHAR(32),
                problem_id VARCHAR(32),
                course_id VARCHAR(32),
                time_id INT,
                is_correct BOOLEAN,
                time_spent_seconds INT COMMENT '答题用时（秒）',
                attempt_count INT DEFAULT 1,
                score_obtained INT DEFAULT 0,
                FOREIGN KEY (student_id) REFERENCES dim_student(student_id),
                FOREIGN KEY (problem_id) REFERENCES dim_problem(problem_id),
                FOREIGN KEY (course_id) REFERENCES dim_course(course_id),
                FOREIGN KEY (time_id) REFERENCES dim_time(time_id)
            )
        """)
    def etl_daily(self, target_date: str):
        """每日 ETL 任务
        从业务数据库（OLTP）抽取数据，转换后加载到数据仓库（OLAP）。
        """
        # 1. 同步维度表（增量更新）
        self._sync_new_students(target_date)
        self._sync_new_problems(target_date)
        # 2. 转换并加载事实数据
        self._load_behavior_facts(target_date)
    def _load_behavior_facts(self, date_str: str):
        """加载学习行为事实数据"""
        # 从业务日志表中提取当天的学习行为
        sql = """
        INSERT INTO fact_learning_behavior
            (student_id, problem_id, course_id, time_id,
             is_correct, time_spent_seconds, attempt_count)
        SELECT
            l.student_id,
            l.problem_id,
            l.course_id,
            -- time_id 计算：YYYYMMDDHH 格式
            CAST(CONCAT(
                DATE_FORMAT(l.event_time, '%Y%m%d'),
                LPAD(HOUR(l.event_time), 2, '0')
            ) AS INT) AS time_id,
            l.is_correct,
            l.time_spent_seconds,
            1
        FROM learning_logs l
        WHERE DATE(l.event_time) = %s
          AND NOT EXISTS (
              -- 去重：同一学生的同一道题当天只保留一条
              SELECT 1 FROM fact_learning_behavior f
              WHERE f.student_id = l.student_id
                AND f.problem_id = l.problem_id
                AND f.time_id = time_id
          )
        """
        self.db.execute(sql, [date_str])
    # ========== 多维分析查询 ==========
    def analyze_weakness_by_dimension(self, course_id: str,
                                     start_date: str,
                                     end_date: str) -> dict:
        """多维度薄弱点分析
        按知识点 × 学生群体进行交叉分析，
        找出哪些学生群体在哪些知识点上最薄弱。
        """
        sql = """
        SELECT
            ds.grade,
            dp.knowledge_point,
            COUNT(*) as total_attempts,
            SUM(CASE WHEN fb.is_correct THEN 1 ELSE 0 END) as correct_count,
            ROUND(
                SUM(CASE WHEN fb.is_correct THEN 1 ELSE 0 END) * 100.0
                / COUNT(*), 1
            ) as accuracy_rate,
            AVG(fb.time_spent_seconds) as avg_time_spent
        FROM fact_learning_behavior fb
        JOIN dim_student ds ON fb.student_id = ds.student_id
        JOIN dim_problem dp ON fb.problem_id = dp.problem_id
        JOIN dim_time dt ON fb.time_id = dt.time_id
        WHERE fb.course_id = %s
          AND dt.full_date BETWEEN %s AND %s
        GROUP BY ds.grade, dp.knowledge_point
        HAVING COUNT(*) >= 10  -- 样本量足够才参与分析
        ORDER BY accuracy_rate ASC
        LIMIT 20
        """
        rows = self.db.fetch_all(sql, [course_id, start_date, end_date])
        return {
            "analysis_period": f"{start_date} ~ {end_date}",
            "weak_areas": [
                {
                    "grade": row["grade"],
                    "knowledge_point": row["knowledge_point"],
                    "accuracy": row["accuracy_rate"],
                    "sample_size": row["total_attempts"],
                    "avg_time": row["avg_time_spent"],
                }
                for row in rows
            ]
        }
    def learning_trend_analysis(self, student_id: str,
                               days: int = 30) -> dict:
        """个人学习趋势分析"""
        sql = """
        SELECT
            dt.full_date as study_date,
            COUNT(*) as daily_problems,
            SUM(CASE WHEN fb.is_correct THEN 1 ELSE 0 END) * 100.0
                / COUNT(*) as daily_accuracy,
            AVG(fb.time_spent_seconds) as avg_time_per_problem
        FROM fact_learning_behavior fb
        JOIN dim_time dt ON fb.time_id = dt.time_id
        WHERE fb.student_id = %s
          AND dt.full_date >= DATE_SUB(CURDATE(), INTERVAL %s DAY)
        GROUP BY dt.full_date
        ORDER BY dt.full_date
        """
        rows = self.db.fetch_all(sql, [student_id, days])
        return {
            "student_id": student_id,
            "days_analyzed": len(rows),
            "trend": [
                {
                    "date": str(row["study_date"]),
                    "problems_solved": int(row["daily_problems"]),
                    "accuracy": round(float(row["daily_accuracy"]), 1),
                    "avg_time": round(float(row["avg_time_per_problem"]), 1),
                }
                for row in rows
            ]
        }

四、边界分析与架构权衡

OLTP vs OLAP

为什么需要数据仓库？因为 OLTP 的范式化设计在分析查询中需要大量 JOIN，性能极差。数据仓库的星型模型虽然数据冗余，但查询性能极高。

数据延迟的接受度

教育数据分析通常不需要实时——T+1（昨天的数据今天分析）对大多数场景已经足够。对于需要实时的场景（如课堂上的即时反馈），可以增加实时聚合层。

五、总结

教育数据仓库的价值在于让"学生的学习行为"从零散日志变为可分析、可比较的结构化数据。星型模型的设计让分析查询变得简单高效。

这个系统的核心经验：

1. 星型模型是分析型查询的最佳数据组织方式
2. 时间维度表虽然看起来"多余"，但让按周/按月/按季度的分析查询简单了太多
3. ETL 是数据仓库持续运行的核心——数据不流动，仓库就是死的

对于后端工程师来说，理解 OLAP 和星型模型是迈向"数据驱动"的关键一步。