Elasticsearch 无法将来自 pymongo 的日期时间字段解析为对象答案

【问题标题】：Elasticsearch fails in parsing datetime field coming from pymongo as objectElasticsearch 无法将来自 pymongo 的日期时间字段解析为对象
【发布时间】：2015-07-28 13:54:13
【问题描述】：

我正在尝试使用 pymongo 和 Python 客户端 elasticsearch 将数据从 mongoDB 流式传输到 Elasticsearch。

我已经设置了一个映射，我这里报出感兴趣的领域相关的sn-p：

"updated_at": { “类型”：“日期”， “格式”：“日期可选时间” }

我的脚本使用 pymongo 从 MongoDB 中抓取每个文档，并尝试将其索引到 Elasticsearch 中

from elasticsearch import Elasticsearch
from pymongo import MongoClient

mongo_client = MongoClient('localhost', 27017)
es_client = Elasticsearch(hosts=[{"host": "localhost", "port": 9200}])
db = mongo_client['my_db']
collection = db['my_collection']

for doc in collection.find():
    es_client.index(
         index='index_name', 
         doc_type='my_type', 
         id=str(doc['_id']), 
         body=json.dumps(doc, default=json_util.default)
    )

我在运行它时遇到的问题是：

elasticsearch.exceptions.RequestError: TransportError(400, u'MapperParsingException[failed to parse [updated_at]]; 嵌套: ElasticsearchIllegalArgumentException[unknown property [$date]]; ')

我认为问题的根源在于 pymongo 将字段 updated_at 序列化为 datetime.datetime 对象，如果我在 for 循环中打印文档，我可以看到：

u'updated_at': datetime.datetime(2014, 8, 31, 17, 18, 13, 17000)

这与 Elasticsearch 寻找映射中指定的 date 类型的对象相冲突。

有什么办法解决这个问题吗？

【问题讨论】：

标签： python mongodb elasticsearch pymongo

【解决方案1】：

您走在正确的道路上，您的 Python datetime 需要序列化为 ISO 8601-compliant 日期字符串。因此，您需要在您的 json.dumps() 通话中添加一个 CustomEncoder。首先，将您的CustomEncoder 声明为JSONEncoder 的子类，它将处理datetime 和time 属性的转换，但将其余部分委托给它的超类：

class CustomEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime):
            return obj.strftime('%Y-%m-%dT%H:%M:%S%z')
        if isinstance(obj, time):
            return obj.strftime('%H:%M:%S')
        if hasattr(obj, 'to_json'):
            return obj.to_json()
        return super(CustomEncoder, self).default(obj)

然后你可以在你的json.dumps调用中使用它，像这样：

...
body=json.dumps(doc, default=json_util.default, cls=CustomEncoder)
...

【讨论】：

这样解决了 datetime 对象的序列化：body 中的字段现在是 "updated_at": {"$date": 1409505493017}。但是，我得到了同样的错误，所以肯定还有其他问题。

【解决方案2】：

我猜你的问题是你正在使用

body=json.dumps(doc, default=json_util.default)

但你应该使用

body=doc

这样做对我有用，因为似乎 elasticsearch 正在处理将字典别名化为 JSON 文档（当然，假设 doc 是字典，我猜是的）。

至少在我使用的弹性搜索版本 (2.x) 中，datetime.datetime 的别名正确，不需要映射。例如，这对我有用：

doc = {"updated_on": datetime.now(timezone.utc)}
res = es.index(index=es_index, doc_type='my_type',
               id=1, body=doc)

并且被 Kibana 识别为日期。

【讨论】：

【解决方案3】：

你可以使用：

from elasticsearch_dsl.serializer import serializer

serializer.dumps(your_dict)

将your_dict 替换为您的Document().prepare() 或document.to_dict()

【讨论】：