咱们先把这个问题拆开来捋——相同代码在预发布（staging）跑的好好的，到生产（production）就炸锅，还报['created']键不存在的错误，这种情况90%是环境配置、ES集群状态或数据差异导致的，和代码本身关系不大，咱们一步步来排查：

一、先解决眼前的['created']键错误 #

这个错误本质是你的代码在尝试从Elasticsearch的索引响应里取response['created']，但生产环境的ES返回的响应结构里根本没有这个键。常见原因有两个：

1. Elasticsearch版本差异 #

不同ES版本的索引响应结构可能不一样：

• 比如ES 7.x及以前，创建文档成功会返回{"created": true}
• 而ES 8.x开始，响应改成了{"result": "created"}

你先分别在staging和prod环境执行这条命令，确认ES版本：

curl -XGET 'http://<你的ES地址>:<端口>'

如果版本不一样，那就是核心问题了——代码写的是兼容旧版本的逻辑，到新版本ES就找不到键了。修复方式也很简单，把取键的逻辑改成兼容写法：

# 原来的写法（可能报错）
if response['created']:
    ...

# 改成兼容版本
is_created = response.get('created') or response.get('result') == 'created'
if is_created:
    ...

2. 索引操作实际失败了 #

如果ES版本一致，那可能是生产环境的索引操作本身就失败了，返回的是错误响应（比如字段类型不匹配、磁盘满了），自然没有created键。这时候你需要在代码里加日志打印完整的响应内容，比如在索引用户的地方：

# 索引用户的代码里添加日志
import logging
logger = logging.getLogger(__name__)

# 执行索引后打印响应
response = es.index(index='users', document=user_data)
logger.info(f"ES索引响应: {response}")

看生产环境的日志里，这个响应到底是成功还是失败，失败的话会有error字段告诉你具体原因。

二、排查staging和prod的环境差异 #

既然代码完全一致，那重点查环境里不一样的地方：

1. ES客户端版本差异 #

检查两个环境的elasticsearch-py（Django用的ES客户端）版本是否一致，执行：

# 在staging和prod分别跑
pip show elasticsearch

如果版本不同，比如staging用7.x，prod用8.x，客户端API的返回结构也会不一样，导致取created键失败。这种情况要么统一客户端版本，要么修改代码兼容。

2. 索引映射的细微差异 #

你已经贴了两个环境的映射，一定要逐字段对比：

• 有没有字段类型不一致？比如staging里某个字段是text，prod里是keyword？
• 有没有字段的null设置不一样？比如staging允许null，prod不允许？
• 有没有prod的映射里缺失某个用户模型的字段？

如果映射有差异，可能导致索引时ES返回异常响应，没有created键。解决方式是把staging的映射同步到prod，或者重新创建prod的索引。

3. ES集群状态问题 #

生产环境的ES集群可能有资源不足或故障：

• 执行这条命令看集群健康状态：

curl -XGET 'http://<你的ES地址>:<端口>/_cluster/health'

如果状态是red或yellow，说明有分片未分配、节点离线或磁盘空间不足的问题，这会直接影响索引操作。需要先修复ES集群的健康问题。

4. 生产环境的用户数据异常 #

预发布环境的测试数据可能比较干净，但生产环境的用户数据可能有异常：

• 比如某个用户的字段值是None，但映射里设置了"null": false
• 比如某个字符串字段超长，超过了映射里的ignore_above设置
• 比如日期字段格式不符合ES要求

你可以在生产环境的Django shell里，单独索引一个用户测试：

from myapp.models import User
from django_elasticsearch_dsl import Index

# 假设你的用户索引是UserIndex
user_index = Index('users')
user = User.objects.first()  # 先拿第一个用户测试
response = user_index.update_object(user)
print(response)

如果单个用户索引失败，就能定位到是数据问题；如果单个成功，那可能是批量索引时的迭代器问题——比如生产环境用户量太大，迭代器在处理到某个用户时出现了数据异常，这时候可以分段批量索引，逐步排查异常数据。

三、关于迭代器的疑似问题 #

如果是批量索引时的迭代器问题，你可以检查：

• 代码里是不是用了User.objects.all()这种大查询集，在生产环境因为数据量太大导致迭代时内存不足或超时？
• 有没有用iterator()方法？比如User.objects.all().iterator()，如果迭代过程中ES连接超时，也会返回异常响应。

可以改成分批处理，比如：

from django.core.paginator import Paginator

users = User.objects.all()
paginator = Paginator(users, 100)  # 每次处理100个用户
for page in paginator.page_range:
    page_users = paginator.page(page)
    for user in page_users:
        # 索引用户
        ...

这样能避免一次性加载大量数据导致的问题。

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