优化Django基于日期子集的聚合查询性能问题

问题背景 #

需要对SQLite中存储的约300万行竞赛Run表，按本年度、上一年度、终身三个时间维度，为40000名参赛者计算多维度汇总统计，最终存储到Report模型中。当前单参赛者聚合耗时约5秒，总耗时不可接受，已尝试拆分过滤聚合、按Class拆分聚合均无明显提升。

模型结构 #

class Competitor(models.Model):
    # ... ID是此处唯一重要的字段

class Venue(models.Model):
    # ... ID是此处唯一重要的字段

class Division(models.Model):
    venue = models.ForeignKey(Venue)
    # ...

class Level(models.Model):
    division = models.ForeignKey(Division)
    # ...

class Class(models.TextChoices):
    STANDARD = "Standard", _("Standard")
    JWW = "JWW", _("JWW")

class Run(models.Model):
    competitor_id = models.ForeignKey(Competitor, related_name="runs", db_index=True)
    date = models.DateField(verbose_name="Date created", db_index=True)
    MACH = models.IntegerField(..., db_index=True)
    PACH = models.IntegerField(..., db_index=True)
    yps = models.FloatField(..., db_index=True)
    score = models.IntegerField(..., db_index=True)
    qualified = models.BooleanField(..., db_index=True)
    division = models.ForeignKey(Division, db_index=True)
    level = models.ForeignKey(Level, db_index=True)
    cls = models.CharField(max_length=..., choices=Class.choices)
    # ... 其他无关字段

class CurrPrevLifetime(models.Model):
    curr_yr = models.FloatField(default=0)
    prev_yr = models.FloatField(default=0)
    lifetime = models.FloatField(default=0)

class Report(models.Model):
    # 根据需要关联多个CurrPrevLifetime实例，存储不同统计项
    ... = models.ForeignKey(CurrPrevLifetime, related_name=...)

当前聚合实现（性能瓶颈） #

curr_yr = Q(date__year=datetime.date.today().year)
prev_yr = Q(date__year=datetime.date.today().year-1)

JWW = Q(cls=Class.JWW)
standard = Q(cls=Class.STANDARD)

aggregates = {
    "curr_yr_max_standard_MACH": Max("MACH", filter=curr_yr & standard),
    "curr_yr_max_standard_PACH": Max("PACH", filter=curr_yr & standard),
    "curr_yr_average_yps_standard": Avg("yps", filter=curr_yr & standard),
    "curr_yr_max_yps_standard": Max("yps", filter=curr_yr & standard),
    "curr_yr_max_JWW_MACH": Max("MACH", filter=curr_yr & JWW),
    "curr_yr_max_JWW_PACH": Max("PACH", filter=curr_yr & JWW),
    "curr_yr_average_yps_JWW": Avg("yps", filter=curr_yr & JWW),
    "curr_yr_max_yps_JWW": Max("yps", filter=curr_yr & JWW),
    "curr_yr_MACH_points": Sum("MACH", filter=curr_yr),
    "curr_yr_PACH_points": Sum("PACH", filter=curr_yr),

    "prev_yr_max_standard_MACH": Max("MACH", filter=prev_yr & standard),
    "prev_yr_max_standard_PACH": Max("PACH", filter=prev_yr & standard),
    "prev_yr_average_yps_standard": Avg("yps", filter=prev_yr & standard),
    "prev_yr_max_yps_standard": Max("yps", filter=prev_yr & standard),
    "prev_yr_max_JWW_MACH": Max("MACH", filter=prev_yr & JWW),
    "prev_yr_max_JWW_PACH": Max("PACH", filter=prev_yr & JWW),
    "prev_yr_average_yps_JWW": Avg("yps", filter=prev_yr & JWW),
    "prev_yr_max_yps_JWW": Max("yps", filter=prev_yr & JWW),
    "prev_yr_MACH_points": Sum("MACH", filter=prev_yr),
    "prev_yr_PACH_points": Sum("PACH", filter=prev_yr),

    "lifetime_max_standard_MACH": Max("MACH", filter=standard),
    "lifetime_max_standard_PACH": Max("PACH", filter=standard),
    "lifetime_average_yps_standard": Avg("yps", filter=standard),
    "lifetime_max_yps_standard": Max("yps", filter=standard),
    "lifetime_max_JWW_MACH": Max("MACH", filter=JWW),
    "lifetime_max_JWW_PACH": Max("PACH", filter=JWW),
    "lifetime_average_yps_JWW": Avg("yps", filter=JWW),
    "lifetime_max_yps_JWW": Max("yps", filter=JWW),
    "lifetime_MACH_points": Sum("MACH"),
    "lifetime_PACH_points": Sum("PACH"),
}
competitor.runs.aggregate(**aggregates)

优化方案 #

1. 批量聚合所有参赛者，避免单条查询 #

当前方案对每个参赛者单独执行聚合，40000次查询的开销极大。改为一次性对所有参赛者分组聚合，直接从Run表计算所有参赛者的统计值，再批量写入Report。

示例代码：

from django.db.models import Case, When, Value, IntegerField
from django.db.models import Max, Avg, Sum

current_year = datetime.date.today().year
prev_year = current_year - 1

# 一次性分组计算所有参赛者的统计数据
aggregated_data = Run.objects.values('competitor_id').annotate(
    # 本年度Standard统计
    curr_yr_max_standard_MACH=Max(Case(When(cls=Class.STANDARD, date__year=current_year, then='MACH'))),
    curr_yr_max_standard_PACH=Max(Case(When(cls=Class.STANDARD, date__year=current_year, then='PACH'))),
    curr_yr_avg_standard_yps=Avg(Case(When(cls=Class.STANDARD, date__year=current_year, then='yps'))),
    curr_yr_max_standard_yps=Max(Case(When(cls=Class.STANDARD, date__year=current_year, then='yps'))),
    # 本年度JWW统计
    curr_yr_max_JWW_MACH=Max(Case(When(cls=Class.JWW, date__year=current_year, then='MACH'))),
    curr_yr_max_JWW_PACH=Max(Case(When(cls=Class.JWW, date__year=current_year, then='PACH'))),
    curr_yr_avg_JWW_yps=Avg(Case(When(cls=Class.JWW, date__year=current_year, then='yps'))),
    curr_yr_max_JWW_yps=Max(Case(When(cls=Class.JWW, date__year=current_year, then='yps'))),
    # 本年度总点数
    curr_yr_MACH_points=Sum(Case(When(date__year=current_year, then='MACH'), default=0)),
    curr_yr_PACH_points=Sum(Case(When(date__year=current_year, then='PACH'), default=0)),
    
    # 上一年度统计（结构同上）
    prev_yr_max_standard_MACH=Max(Case(When(cls=Class.STANDARD, date__year=prev_year, then='MACH'))),
    prev_yr_max_standard_PACH=Max(Case(When(cls=Class.STANDARD, date__year=prev_year, then='PACH'))),
    prev_yr_avg_standard_yps=Avg(Case(When(cls=Class.STANDARD, date__year=prev_year, then='yps'))),
    prev_yr_max_standard_yps=Max(Case(When(cls=Class.STANDARD, date__year=prev_year, then='yps'))),
    prev_yr_max_JWW_MACH=Max(Case(When(cls=Class.JWW, date__year=prev_year, then='MACH'))),
    prev_yr_max_JWW_PACH=Max(Case(When(cls=Class.JWW, date__year=prev_year, then='PACH'))),
    prev_yr_avg_JWW_yps=Avg(Case(When(cls=Class.JWW, date__year=prev_year, then='yps'))),
    prev_yr_max_JWW_yps=Max(Case(When(cls=Class.JWW, date__year=prev_year, then='yps'))),
    prev_yr_MACH_points=Sum(Case(When(date__year=prev_year, then='MACH'), default=0)),
    prev_yr_PACH_points=Sum(Case(When(date__year=prev_year, then='PACH'), default=0)),
    
    # 终身统计
    lifetime_max_standard_MACH=Max(Case(When(cls=Class.STANDARD, then='MACH'))),
    lifetime_max_standard_PACH=Max(Case(When(cls=Class.STANDARD, then='PACH'))),
    lifetime_avg_standard_yps=Avg(Case(When(cls=Class.STANDARD, then='yps'))),
    lifetime_max_standard_yps=Max(Case(When(cls=Class.STANDARD, then='yps'))),
    lifetime_max_JWW_MACH=Max(Case(When(cls=Class.JWW, then='MACH'))),
    lifetime_max_JWW_PACH=Max(Case(When(cls=Class.JWW, then='PACH'))),
    lifetime_avg_JWW_yps=Avg(Case(When(cls=Class.JWW, then='yps'))),
    lifetime_max_JWW_yps=Max(Case(When(cls=Class.JWW, then='yps'))),
    lifetime_MACH_points=Sum('MACH'),
    lifetime_PACH_points=Sum('PACH'),
)

# 批量创建统计实例和报告
curr_prev_lifetime_objects = []
report_objects = []

for item in aggregated_data:
    # 拆分统计项到CurrPrevLifetime实例
    mach_standard = CurrPrevLifetime(
        curr_yr=item['curr_yr_max_standard_MACH'] or 0,
        prev_yr=item['prev_yr_max_standard_MACH'] or 0,
        lifetime=item['lifetime_max_standard_MACH'] or 0
    )
    mach_jww = CurrPrevLifetime(
        curr_yr=item['curr_yr_max_JWW_MACH'] or 0,
        prev_yr=item['prev_yr_max_JWW_MACH'] or 0,
        lifetime=item['lifetime_max_JWW_MACH'] or 0
    )
    # ... 其他统计项同理创建实例
    curr_prev_lifetime_objects.extend([mach_standard, mach_jww])

# 批量写入数据库
CurrPrevLifetime.objects.bulk_create(curr_prev_lifetime_objects, batch_size=1000)

# 关联到Report模型（根据实际结构调整）
for item in aggregated_data:
    report = Report(
        competitor_id=item['competitor_id'],
        # 关联已创建的CurrPrevLifetime实例
        mach_standard_max=...,
        mach_jww_max=...,
        # ... 其他字段
    )
    report_objects.append(report)

Report.objects.bulk_create(report_objects, batch_size=1000)

2. 添加复合索引，加速分组聚合 #

SQLite的聚合查询依赖复合索引优化，添加以下索引：

class Run(models.Model):
    # ... 原有字段
    class Meta:
        indexes = [
            # 覆盖参赛者+日期+Class的过滤分组
            models.Index(fields=['competitor_id', 'date', 'cls'], name='run_competitor_date_cls_idx'),
            # 覆盖参赛者+Class+聚合字段的索引
            models.Index(fields=['competitor_id', 'cls', 'MACH', 'PACH', 'yps'], name='run_competitor_cls_agg_idx'),
        ]

添加后运行migrate，并使用EXPLAIN QUERY PLAN验证索引是否生效。

3. 预计算缓存，避免重复计算 #

如果统计无需实时更新，可定期（如每日凌晨）执行批量聚合，将结果存储到Report表。后续查询直接读取Report数据，无需重复计算。

4. 优化SQLite配置 #

调整SQLite参数提升性能：

from django.db.backends.signals import connection_created
from django.dispatch import receiver

@receiver(connection_created)
def configure_sqlite(sender, connection, **kwargs):
    if connection.vendor == 'sqlite':
        cursor = connection.cursor()
        cursor.execute('PRAGMA journal_mode=WAL;')  # 减少写锁竞争
        cursor.execute('PRAGMA cache_size=-20000;')  # 增大缓存至80MB（每页4KB）
        cursor.execute('PRAGMA optimize;')  # 自动优化查询计划

5. 反规范化模型结构（可选） #

若性能仍不达标，可将统计数据反规范化，在写入Run时实时更新统计值：

• 在Competitor模型添加终身统计字段，每次新增Run时更新。
• 创建CompetitorYearlyStats模型，按参赛者+年份存储年度统计，写入Run时同步更新对应年份数据。

这种方式将聚合开销分散到写入阶段，读取性能最优，但需维护数据一致性。

内容的提问来源于stack exchange，提问作者Quontas