Django避免模型重复：多量级气象数据架构优化问询

模型层：消除重复字段定义 #

• 定义抽象基类，将所有模型共有的datetime、location_id、average字段抽离，所有气象要素的各时间粒度模型继承该基类：

from django.db import models

class BaseMeteorologicalModel(models.Model):
    datetime = models.DateTimeField()
    location_id = models.IntegerField(db_index=True)
    average = models.FloatField(null=True)

    class Meta:
        abstract = True
        indexes = [
            models.Index(fields=['location_id', 'datetime']),  # 预建联合索引提升查询效率
        ]

• 批量生成具体模型：针对20余种气象要素，通过循环动态创建raw/hourly/daily/monthly四个粒度的模型，避免手动重复编写：

# 定义所有气象要素和时间粒度
METEOROLOGICAL_ELEMENTS = ['temperature', 'precipitation', 'wind_speed', ...]
TIME_GRANULARITIES = ['raw', 'hourly', 'daily', 'monthly']

for element in METEOROLOGICAL_ELEMENTS:
    for granularity in TIME_GRANULARITIES:
        # 动态生成模型类名和表名
        model_name = f"{element.capitalize()}{granularity.capitalize()}"
        table_name = f"{element}_{granularity}"
        # 创建模型元类
        meta_class = type('Meta', (), {'db_table': table_name})
        # 动态生成模型类并注册到全局
        model_class = type(
            model_name,
            (BaseMeteorologicalModel,),
            {'__module__': __name__, 'Meta': meta_class}
        )
        globals()[model_name] = model_class

过滤器与视图：复用通用逻辑 #

• 通用过滤器集：创建基于基类模型的通用过滤器，支持时间范围、地点ID筛选，使用时仅需传入具体模型：

import django_filters

class MeteorologicalFilterSet(django_filters.FilterSet):
    start_datetime = django_filters.DateTimeFilter(field_name='datetime', lookup_expr='gte')
    end_datetime = django_filters.DateTimeFilter(field_name='datetime', lookup_expr='lte')
    location_id = django_filters.NumberFilter()

    class Meta:
        model = BaseMeteorologicalModel
        fields = ['start_datetime', 'end_datetime', 'location_id']

# 快捷获取对应模型的过滤器
def get_meteorological_filter(model_class, request_data):
    return MeteorologicalFilterSet(data=request_data, queryset=model_class.objects.all())

• 通用视图集：基于Django REST Framework的ReadOnlyModelViewSet封装通用视图，通过动态指定模型类实现复用：

from rest_framework import viewsets

class MeteorologicalViewSet(viewsets.ReadOnlyModelViewSet):
    filterset_class = MeteorologicalFilterSet
    model_class = None

    def get_queryset(self):
        return self.model_class.objects.all().order_by('datetime')

# 快速注册各模型视图
temperature_raw_viewset = type(
    'TemperatureRawViewSet',
    (MeteorologicalViewSet,),
    {'model_class': TemperatureRaw}
)

数据处理：统一清理与聚合逻辑 #

• 通用数据清理函数：封装原始数据的缺失值、无效值校验逻辑，适配所有气象要素：

def clean_raw_data(raw_queryset, element_type):
    # 根据要素类型定义校验规则，比如温度不能低于绝对零度
    valid_rules = {
        'temperature': lambda qs: qs.filter(average__gte=-273.15),
        'precipitation': lambda qs: qs.filter(average__gte=0),
        # 其他要素规则...
    }
    # 先过滤空值，再应用要素专属规则
    cleaned = raw_queryset.filter(average__isnull=False)
    return valid_rules[element_type](cleaned)

• 通用聚合函数：根据时间粒度自动生成聚合逻辑，批量插入目标模型：

from django.db.models import Avg
from datetime import timedelta

def aggregate_to_granularity(raw_model, target_model, granularity, element_type):
    cleaned_data = clean_raw_data(raw_model.objects.all(), element_type)
    # 映射时间粒度到Django的Trunc参数
    trunc_map = {
        'hourly': 'hour',
        'daily': 'day',
        'monthly': 'month'
    }
    # 按地点和时间粒度分组聚合
    aggregated = cleaned_data.annotate(
        grouped_datetime=models.Trunc('datetime', trunc_map[granularity], output_field=models.DateTimeField())
    ).values('grouped_datetime', 'location_id').annotate(
        avg_value=Avg('average')
    )
    # 批量插入目标表，提升效率
    target_objects = [
        target_model(
            datetime=item['grouped_datetime'],
            location_id=item['location_id'],
            average=item['avg_value']
        ) for item in aggregated
    ]
    target_model.objects.bulk_create(target_objects)

调用示例：

# 生成温度小时级聚合数据
aggregate_to_granularity(TemperatureRaw, TemperatureHourly, 'hourly', 'temperature')

额外优化建议 #

• 用Celery等任务队列异步执行数据清理与聚合任务，避免阻塞主线程，适配持续增长的大数据量场景。
• 将气象要素、时间粒度、校验规则等配置统一存入常量类或配置文件，便于后续扩展和修改。

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