Django海量商品数据统计：实时计算vs定时任务最佳实践咨询

Hey Jasper, 我刚好在之前的项目里处理过类似的百万级电商爬虫数据统计问题，给你分享几个经过实践验证的最佳思路，应该能帮你解决延迟和后续扩展的问题：

其实大部分用户场景里，不是所有统计都需要绝对实时：

• 比如商品当前价格、最新卖家信息这种，必须实时，直接从主表查询就行；
• 但价格走势折线图、价格分布、爬取报告统计这类，用户完全能接受1-2小时的延迟，这部分就交给定时预计算，绝对能解决性能问题。

具体落地方式：

• 新增一个专门用于存储预计算结果的模型，比如ProductStats，把需要展示的统计字段（比如价格走势数据、平均价、分布区间）都存在这里；
• 用Celery+Beat（或者系统CRON）在独立服务器上定时跑计算任务，把结果写入这个模型。前端直接查询这个模型，速度和查普通数据没区别。

举个简单的代码示例：

# models.py
from django.db import models
from django.contrib.postgres.fields import JSONField

class ProductStats(models.Model):
    product = models.ForeignKey("Product", on_delete=models.CASCADE, related_name="stats")
    price_trend = JSONField()  # 存时间-价格的列表，供折线图用
    avg_price = models.DecimalField(max_digits=10, decimal_places=2)
    price_distribution = JSONField()  # 存价格区间的统计结果
    last_updated = models.DateTimeField(auto_now=True)

# tasks.py
from celery import shared_task
from django.db.models import Avg
from .models import Product, ProductStats, ProductPriceHistory

def calculate_price_distribution(prices):
    # 自定义计算价格分布的逻辑，比如分10个区间统计数量
    min_price = min(prices)
    max_price = max(prices)
    interval = (max_price - min_price) / 10
    dist = []
    for i in range(10):
        lower = min_price + interval*i
        upper = lower + interval
        count = len([p for p in prices if lower <= p < upper])
        dist.append({"interval": f"{lower:.2f}-{upper:.2f}", "count": count})
    return dist

@shared_task
def update_product_stats(product_id=None):
    # 可以指定单个商品，或者批量更新所有有新数据的商品
    products = Product.objects.all() if not product_id else [Product.objects.get(id=product_id)]
    for product in products:
        price_history = ProductPriceHistory.objects.filter(product=product).order_by("crawl_time")
        if not price_history.exists():
            continue
        # 用ORM聚合函数让数据库做计算，比Python循环快N倍
        avg_price = price_history.aggregate(Avg("price"))["price__avg"]
        # 生成折线图数据（只存最近30天，减少存储）
        trend_data = [
            {"time": item.crawl_time.isoformat(), "price": float(item.price)}
            for item in price_history.order_by("crawl_time")[-30:]
        ]
        # 计算价格分布
        price_dist = calculate_price_distribution(price_history.values_list("price", flat=True))
        # 更新或创建统计记录
        ProductStats.objects.update_or_create(
            product=product,
            defaults={
                "price_trend": trend_data,
                "avg_price": avg_price,
                "price_distribution": price_dist
            }
        )

定时任务可以配置成每小时跑一次，甚至可以优化成增量计算：只更新最近有价格变动的商品，减少不必要的计算量。

如果某些场景确实需要实时统计（比如用户刚手动触发爬取后要查看最新结果），也不用慌，做这几个优化就能撑住百万级数据：

• 加联合索引：比如给ProductPriceHistory的product_id和crawl_time加联合索引，查询历史价格的速度会提升一个量级；
• 用数据库聚合：尽量用Django ORM的Avg、Count、Sum这些聚合函数，让数据库帮你做计算，而不是把所有数据拉到Python里循环处理；
• 加缓存层：用Redis缓存高频查询的实时结果，比如热门商品的实时价格，缓存5-10分钟，过期后再重新计算，能大幅减少数据库压力。

后续要加的回归分析、近邻分类这类计算，绝对不能放到视图里跑——百万级数据下，一次计算可能要几分钟甚至更久。正确的做法是：

• 把这些任务放到独立的机器学习服务里，定时从数据库拉取批量数据，训练模型、计算结果；
• 把计算好的结果（比如商品的分类标签、价格预测值）存回Django的模型或者专门的存储（比如Redis），前端直接查询结果即可；
• 如果需要增量更新（比如新爬取的商品要实时分类），可以用消息队列（比如RabbitMQ），当新商品入库时发送消息给ML服务，触发增量计算。

百万级的时序数据（比如价格历史），用MySQL这类关系型数据库查询效率会越来越低，可以考虑：

• 把历史价格数据放到时序数据库（比如InfluxDB），专门处理时间序列数据，查询时间范围的走势会快很多；
• 用Elasticsearch做统计分析，它的聚合功能非常强大，适合做价格分布、卖家信息统计这类复杂分析，而且支持全文检索，后续扩展搜索功能也方便。

即使后端数据快，前端渲染大数据量的折线图也可能卡顿，所以可以做：

• 分页加载数据：比如默认只展示最近30天的价格走势，用户需要更早的数据再点击加载；
• 用Canvas绘制图表：比如ECharts这类基于Canvas的图表库，比DOM渲染快很多，能轻松处理上万条数据的折线图。

总结一下：优先用预计算+缓存处理大部分统计场景，平衡延迟和性能；实时统计做索引和聚合优化；复杂ML任务完全离线处理；数据库根据数据类型拆分。这样既能解决当前的性能问题，也能支撑后续的扩展需求。

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