基于Django 3.2.13与MySQL 5.7实现DB Sharding的技术问询

针对你的业务场景（按用户查询交易数据、按时间区间查询全量数据），以下是适配的分片思路、实用参考和代码实现：

核心分片策略适配 #

• 场景1（按用户查海量交易）：采用用户ID哈希分片，将同一用户的所有交易数据路由到同一分片，避免跨库查询，提升单用户查询效率。
• 场景2（按时间区间查全量数据）：结合时间范围分片（按年月拆分），查询时遍历对应时间区间的分片并行查询后聚合结果，平衡跨分片查询的性能开销。

实用参考内容 #

• Django官方数据库路由（Database Routers）机制：这是实现多数据库路由的核心基础，通过自定义路由类控制读写操作的库分配逻辑。
• 自定义模型管理器封装：扩展Django模型管理器，将分片查询逻辑封装为易用的方法，屏蔽底层分片细节。
• MySQL分片最佳实践：优先选择查询占比最高的字段作为分片键；避免跨分片事务；针对时间分片定期归档冷数据，降低单库数据量。

代码示例 #

1. 配置多分片数据库（settings.py） #

# settings.py
DATABASES = {
    'default': {},  # 默认库留空，由路由控制
    # 用户哈希分片库示例
    'shard_user_0': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': 'trade_shard_0',
        'USER': 'db_user',
        'PASSWORD': 'db_pass',
        'HOST': 'shard-host-0',
        'PORT': '3306',
    },
    'shard_user_1': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': 'trade_shard_1',
        'USER': 'db_user',
        'PASSWORD': 'db_pass',
        'HOST': 'shard-host-1',
        'PORT': '3306',
    },
    # 时间范围分片库示例（按年月命名）
    'shard_time_202401': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': 'trade_time_202401',
        'USER': 'db_user',
        'PASSWORD': 'db_pass',
        'HOST': 'shard-time-host-0',
        'PORT': '3306',
    },
    'shard_time_202402': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': 'trade_time_202402',
        'USER': 'db_user',
        'PASSWORD': 'db_pass',
        'HOST': 'shard-time-host-1',
        'PORT': '3306',
    },
    # 更多分片库按需添加
}

# 启用自定义数据库路由
DATABASE_ROUTERS = ['myapp.routers.TradeShardRouter']

2. 自定义分片路由类（routers.py） #

# myapp/routers.py
import hashlib
from django.conf import settings

class TradeShardRouter:
    def db_for_read(self, model, **hints):
        # 处理用户ID分片的读请求
        if model._meta.model_name == 'trade':
            user_id = hints.get('user_id')
            if user_id:
                shard_count = len([k for k in settings.DATABASES if k.startswith('shard_user_')])
                shard_num = int(hashlib.md5(str(user_id).encode()).hexdigest(), 16) % shard_count
                return f'shard_user_{shard_num}'
        return 'default'

    def db_for_write(self, model, **hints):
        # 写入时按用户ID路由到对应分片
        if model._meta.model_name == 'trade':
            user_id = hints.get('user_id')
            if user_id:
                shard_count = len([k for k in settings.DATABASES if k.startswith('shard_user_')])
                shard_num = int(hashlib.md5(str(user_id).encode()).hexdigest(), 16) % shard_count
                return f'shard_user_{shard_num}'
        return 'default'

    def allow_relation(self, obj1, obj2, **hints):
        # 允许同一分片内的模型关联
        return True

    def allow_migrate(self, db, app_label, model_name=None, **hints):
        # 控制Trade模型仅迁移到分片库
        if db.startswith('shard_user_') or db.startswith('shard_time_'):
            return model_name == 'trade'
        return False

3. 自定义模型管理器处理跨分片查询（models.py） #

# myapp/models.py
from django.db import models, connections
from django.db.models import Manager

class TradeManager(Manager):
    def get_by_user(self, user_id):
        """按用户ID查询对应分片的交易数据"""
        shard_name = self._get_user_shard(user_id)
        return self.using(shard_name).filter(user_id=user_id)
    
    def get_by_time_range(self, start_time, end_time):
        """按时间区间遍历对应分片查询并聚合结果"""
        shard_list = self._get_time_shards(start_time, end_time)
        all_results = []
        
        for shard in shard_list:
            with connections[shard].cursor() as cursor:
                cursor.execute("""
                    SELECT id, user_id, trade_time, amount 
                    FROM myapp_trade 
                    WHERE trade_time BETWEEN %s AND %s
                """, [start_time, end_time])
                columns = [col[0] for col in cursor.description]
                all_results.extend([dict(zip(columns, row)) for row in cursor.fetchall()])
        
        return all_results
    
    def _get_user_shard(self, user_id):
        """计算用户ID对应的分片名称"""
        from django.conf import settings
        shard_count = len([k for k in settings.DATABASES if k.startswith('shard_user_')])
        shard_num = int(hashlib.md5(str(user_id).encode()).hexdigest(), 16) % shard_count
        return f'shard_user_{shard_num}'
    
    def _get_time_shards(self, start_time, end_time):
        """生成时间区间对应的分片名称列表"""
        shard_list = []
        current_month = start_time.replace(day=1)
        end_month = end_time.replace(day=1)
        
        while current_month <= end_month:
            shard_name = f'shard_time_{current_month.strftime("%Y%m")}'
            shard_list.append(shard_name)
            
            # 切换到下一个月
            if current_month.month == 12:
                current_month = current_month.replace(year=current_month.year + 1, month=1)
            else:
                current_month = current_month.replace(month=current_month.month + 1)
        
        return shard_list

class Trade(models.Model):
    user_id = models.IntegerField()
    trade_time = models.DateTimeField()
    amount = models.DecimalField(max_digits=10, decimal_places=2)
    # 其他业务字段按需添加

    objects = TradeManager()

关键注意事项 #

• 跨分片事务：Django默认不支持跨库事务，若需事务保证，需避免跨分片写操作，或引入分布式事务组件。
• 数据迁移：已有数据需按分片规则编写脚本迁移到对应分片库，确保数据分布符合预期。
• 监控优化：针对每个分片库单独监控性能指标，定期对时间分片的冷数据进行归档或清理。

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