在Python(Flask)中用Elasticsearch获取最近地理点遇阻求助

我正在跟着Elasticsearch官方教程，在Python（Flask）环境下开发，已实现全文匹配搜索功能，现扩展「获取最近地理位置」功能时，参考geo-distance query文档编写的查询无法正常工作。

相关代码 #

Search类实现 #

from pprint import pprint

from elasticsearch import Elasticsearch
from decouple import config
from managers.home_manager import HomeManager
from schemas.response.home_response import HomeResponseSchema, PinSchema

mapping = {
    "mappings": {
        "properties": {"pin": {"properties": {"location": {"type": "geo_point"}}}}
    }
}


class Search:
    def __init__(self) -> None:
        self.es = Elasticsearch(
            api_key=config("ELASTIC_API_KEY"), cloud_id=config("ELASTIC_CLOUD_ID")
        )
        client_info = self.es.info()
        print("Connected to Elasticsearch!")
        pprint(client_info.body)

    def create_index(self):
        self.es.indices.delete(index="real_estate_homes", ignore_unavailable=True)
        self.es.indices.create(index="real_estate_homes", body=mapping)

    def insert_document(self, document):
        return self.es.index(index="real_estate_homes", body=document)

    def insert_documents(self, documents):
        operations = []
        for document in documents:
            operations.append({"index": {"_index": "real_estate_homes"}})
            operations.append(document)
        return self.es.bulk(operations=operations)

    def reindex_homes(self):
        self.create_index()
        homes = HomeManager.select_all_homes()
        pins = []
        for home in homes:
            pin = {
                "location": {"lat": float(home.latitude), "lon": float(home.longitude)}
            }
            pins.append(pin)
        
        return self.insert_documents(pins)

    def search(self, **query_args):
        return self.es.search(index="real_estate_homes", **query_args)


es = Search()

查询接口代码 #

from flask_restful import Resource

from managers.home_manager import HomeManager
from search import es
from schemas.response.home_response import HomeResponseSchema


class ElasticResource(Resource):
    def get(self, home_id):
        print(home_id)
        home = HomeManager.select_home_by_id(home_id)
        geo_query = es.search(
            body={
                "query": {
                    "bool": {
                        "must": {"match_all": {}},
                        "filter": {
                            "geo_distance": {
                                "distance": "2000km",
                                "pin.location": {"lat": 43, "lon": 27},
                            }
                        },
                    }
                }
            }
        )
        print(geo_query)
        result = es.search(
            query={
                "bool": {
                    "must": {"match": {"city": {"query": home.city}}},
                    "must_not": {"match": {"id": {"query": home.id}}},
                }
            }
        )
        print(result["hits"]["hits"])
        if len(result["hits"]["hits"]) == 0:
            return "No results."
        suggested_homes = [
            suggested_home["_source"] for suggested_home in result["hits"]["hits"]
        ]
        resp_schema = HomeResponseSchema()
        return resp_schema.dump(suggested_homes, many=True), 200

问题根源 #

1. 数据结构与mapping不匹配：索引mapping定义的是pin.location字段，但插入数据时仅存入了location，缺少外层pin结构，导致查询时无法识别目标字段。
2. 地理查询未实际生效：代码中定义了geo_query但未使用，实际返回结果仅基于城市匹配，地理过滤逻辑未参与查询。

修复方案 #

1. 修正数据插入结构 #

修改reindex_homes方法，确保数据结构与mapping一致，同时补充后续查询需要的字段：

def reindex_homes(self):
    self.create_index()
    homes = HomeManager.select_all_homes()
    docs = []
    for home in homes:
        doc = {
            "pin": {
                "location": {"lat": float(home.latitude), "lon": float(home.longitude)}
            },
            "city": home.city,
            "id": home.id
            # 可补充其他需要搜索的字段，如房屋名称、描述等
        }
        docs.append(doc)
    
    return self.insert_documents(docs)

2. 整合地理查询与现有逻辑 #

修改ElasticResource的get方法，将地理距离过滤与城市匹配、排除当前房屋的逻辑结合：

class ElasticResource(Resource):
    def get(self, home_id):
        home = HomeManager.select_home_by_id(home_id)
        # 组合查询：城市匹配 + 地理距离过滤 + 排除当前房屋
        result = es.search(
            query={
                "bool": {
                    "must": {"match": {"city": {"query": home.city}}},
                    "must_not": {"match": {"id": {"query": home.id}}},
                    "filter": {
                        "geo_distance": {
                            "distance": "2000km",
                            "pin.location": {"lat": float(home.latitude), "lon": float(home.longitude)}
                        }
                    }
                }
            }
        )
        hits = result["hits"]["hits"]
        if not hits:
            return "No results.", 200
        suggested_homes = [hit["_source"] for hit in hits]
        resp_schema = HomeResponseSchema()
        return resp_schema.dump(suggested_homes, many=True), 200

3. 重新索引数据 #

调用es.reindex_homes()重新导入数据，确保新的结构和字段生效。

验证步骤 #

• 检查索引mapping是否正确：

  print(es.es.indices.get_mapping(index="real_estate_homes"))
  

• 查看单条文档结构，确认pin.location存在：

  # 替换为实际文档ID
  print(es.es.get(index="real_estate_homes", id="1"))
  

内容的提问来源于stack exchange，提问作者Mr. Terminix