如何在Django中用原生查询关联外部数据库与Django模型并存储结果

核心思路 #

因为涉及Django内部数据库与外部SQL Server的跨库关联查询，Django ORM本身并不支持高效的跨库关联操作，所以最优方案是利用数据库原生跨库查询能力完成数据关联过滤，再通过批量插入将结果写入目标Django模型——这种方式能充分发挥数据库的计算性能，避免Python内存过载，适配百万级数据场景。

具体步骤 #

1. 编写原生跨库查询SQL #

首先确认外部SQL Server与Django数据库的跨库访问权限（若两库在同一SQL Server实例，直接用[库名].[架构名].[表名]即可；不同实例需先配置链接服务器）。修正你提供的SQL语法错误后，示例如下：

SELECT 
    x.product,
    d.client_name
FROM 
    [externalDB].[dbo].[your_external_table] x  -- 替换为实际外部表名和架构
INNER JOIN 
    [django_default_db].[dbo].[appname_djangomodel] d ON x.client_id = d.client_id
WHERE 
    d.client_name = 'foo'

注：django_default_db是你Django默认数据库的名称，appname_djangomodel是对应Django模型的数据库表名（通常为应用名_模型名小写形式）。

2. 执行查询并批量插入结果 #

使用Django配置的外部数据库连接执行查询，通过游标逐批读取结果（避免一次性加载百万数据到内存），再用bulk_create批量写入目标模型：

from django.db import connections
from your_app.models import TargetModel  # 替换为你的目标Django模型

batch_size = 1000  # 可根据服务器内存/数据库性能调整，推荐1000-5000区间
with connections['externalDB'].cursor() as cursor:
    # 执行跨库查询
    cursor.execute("""
        SELECT 
            x.product,
            d.client_name
        FROM 
            [externalDB].[dbo].[your_external_table] x
        INNER JOIN 
            [django_default_db].[dbo].[appname_djangomodel] d ON x.client_id = d.client_id
        WHERE 
            d.client_name = 'foo'
    """)
    # 逐批读取并插入
    while True:
        rows = cursor.fetchmany(batch_size)
        if not rows:
            break
        # 构造目标模型实例列表
        target_instances = [
            TargetModel(
                product=row[0],
                client_name=row[1]
            )
            for row in rows
        ]
        # 批量插入，性能远高于单条save()
        TargetModel.objects.bulk_create(target_instances, batch_size=batch_size)

3. 性能优化建议 #

• 索引优化：为外部表的client_id、Django模型表的client_id和client_name字段创建索引，大幅提升跨库JOIN和WHERE过滤的速度。
• 事务控制：若需要保证数据一致性，可在批量插入时包裹事务：

  from django.db import transaction
  
  with transaction.atomic():
      TargetModel.objects.bulk_create(target_instances, batch_size=batch_size)
  

• 去重处理：如果需要避免重复插入，可在SQL中添加过滤逻辑：

  SELECT 
      x.product,
      d.client_name
  FROM 
      [externalDB].[dbo].[your_external_table] x
  INNER JOIN 
      [django_default_db].[dbo].[appname_djangomodel] d ON x.client_id = d.client_id
  WHERE 
      d.client_name = 'foo'
      AND NOT EXISTS (
          SELECT 1 FROM [django_default_db].[dbo].[appname_targetmodel] t 
          WHERE t.product = x.product AND t.client_name = d.client_name
      )
  

4. 不推荐的备选方案（仅适用于小数据量） #

如果非要用Django ORM实现，只能先拉取两边数据再在Python内存中关联，但这种方式会导致内存占用过高，完全不适合百万级数据：

# 仅小数据量场景使用！
from django.db import connections
from your_app.models import DjangoModel, TargetModel

# 先获取符合条件的client_id集合
client_ids = list(DjangoModel.objects.filter(client_name='foo').values_list('client_id', flat=True))

batch_size = 1000
with connections['externalDB'].cursor() as cursor:
    cursor.execute("""
        SELECT product, client_id 
        FROM [dbo].[your_external_table] 
        WHERE client_id IN %s
    """, (tuple(client_ids),))
    # 预存client_id与client_name的映射，避免重复查询
    client_map = {
        item['client_id']: item['client_name'] 
        for item in DjangoModel.objects.filter(client_id__in=client_ids).values('client_id', 'client_name')
    }
    while True:
        rows = cursor.fetchmany(batch_size)
        if not rows:
            break
        target_instances = [
            TargetModel(product=row[0], client_name=client_map[row[1]])
            for row in rows
        ]
        TargetModel.objects.bulk_create(target_instances, batch_size=batch_size)

总结 #

对于百万级跨库关联查询+批量插入的场景，数据库原生跨库查询+游标逐批读取+Django bulk_create批量插入是性能最高、最稳定的方案，能最大程度利用数据库的计算能力，避免Python内存瓶颈。

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