经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 程序设计 » Elasticsearch » 查看文章
利用Elasticsearch实现地理位置、城市搜索服务
来源：cnblogs　　作者：下午喝什么茶　　时间：2024/7/29 9:31:01　　对本文有异议

最近用到一些简单的地理位置查询接口，基于当前定位获取用户所在位置信息（省市区），然后基于该信息查询当前区域的......提供服务。

然后就自己研究了下GIS，作为一个程序员。自己能不能实现这个功能呢？答案当然是可以。立即开干。

思路：找到数据，写入数据库，利用Elasticsearch强大的搜索能力和丰富发热GIS数据处理能力实现。

GIS相关专业信息参考（bd上找到，还算专业）：程序员GIS入门|前后端都要懂一点的GIS知识

经过一番寻找，“功夫不负有心人”，在网上找到了锐多宝 数据，比较完整。下载下来，格式是shape格式。

第一步：下载数据，从锐多宝下载

 第二步：写python脚本预处理数据：ShapFile 转 GeoJSON，ES处理GeoJSON比较强

  1. import geopandas as gpd

    # 读取 Shapefile
    shapefile_path = 'D:/data/gis/2023年_CTAmap_1.12版/2023年省级/2023年省级.shp'
    gdf = gpd.read_file(shapefile_path)

    # 检查 GeoDataFrame
    print(gdf.head())

    # 如果需要，可以对数据进行预处理，比如过滤、选择特定列等
    # gdf = gdf[['column1', 'column2', 'geometry']]

    # 将 GeoDataFrame 转换为标准的 Pandas DataFrame (如果需要的话)
    df = gdf.drop('geometry', axis=1).join(gdf['geometry'].apply(lambda x: gpd.GeoSeries(x).to_json()))

    # 将 Pandas DataFrame 导出为 JSON 文件
    output_json_path = 'D:/data/gis/2023-province-GeoJSON.gesjson'
    # df.to_json(output_json_path, orient='records')

    # 如果你想保留 GeoJSON 格式，可以直接保存 GeoDataFrame
    gdf.to_file(output_json_path, driver='GeoJSON')

第三步：利用Python脚本将GeoJSON写入Elasticsearch

  1. from elasticsearch import Elasticsearch
  2. from elasticsearch.helpers import bulk
  3. import json
  5. # 连接到 Elasticsearch
  6. es = Elasticsearch("http://localhost:9200")
  8. # 检查连接
  9. if not es.ping():
  10.     raise ValueError("Connection failed")
  12. # 删除旧索引（如果存在）
  13. if es.indices.exists(index="province2023_geoshape_index_001"):
  14.     es.indices.delete(index="province2023_geoshape_index_001")
  16. # 创建索引并定义 Mapping
  17. mapping = {
  18.     "mappings": {
  19.         "properties": {
  20.             "location": {
  21.                 "type": "geo_shape"
  22.             },
  23.             "name": {
  24.                 "type": "text"
  25.             }
  26.         }
  27.     }
  28. }
  30. # 创建索引
  31. es.indices.create(index="province2023_geoshape_index_001", body=mapping)
  33. # 读取 GeoJSON 文件
  34. with open("D:/data/gis/2023-province-GeoJSON.gesjson", "r", encoding="utf-8") as file:
  35.     geojson_data = json.load(file)
  37. # 提取 GeoJSON 特征集合
  38. features = geojson_data.get("features", [])
  40. # 准备数据以供导入
  41. documents = []
  42. for feature in features:
  43.     doc = {
  44.         "location": {
  45.             "type": feature["geometry"]["type"],
  46.             "coordinates": feature["geometry"]["coordinates"]
  47.         }
  48.     }
  49.     if "properties" in feature:
  50.         doc.update(feature["properties"])
  51.     documents.append(doc)
  53. # 定义批量大小
  54. batch_size = 100  # 每次批量导入的数量
  57. # 准备 actions
  58. def generate_actions(documents):
  59.     for doc in documents:
  60.         yield {
  61.             "_index": "province2023_geoshape_index_001",
  62.             "_source": doc
  63.         }
  66. # 分批执行批量导入
  67. for i in range(0, len(documents), batch_size):
  68.     end = min(i + batch_size, len(documents))
  69.     success, _ = bulk(es, generate_actions(documents[i:end]))
  70.     print(f"Bulk {i}-{end} completed, {success} documents indexed.")
  72. print("All data indexed.")

第四步：计算出每条数据的区域的中心点（扩展功能，原始数据只有polygon多边形数据）

  1. from elasticsearch import Elasticsearch
  2. from elasticsearch.helpers import bulk
  3. import json
  4. import ssl
  6. # 连接到 Elasticsearch
  7. es = Elasticsearch("http://localhost:9200")
  9. # 检查连接
  10. if not es.ping():
  11.     raise ValueError("Connection failed")
  13. # 删除旧索引（如果存在）
  14. if es.indices.exists(index="province2023_centroid_geoshape_index_001"):
  15.     es.indices.delete(index="province2023_centroid_geoshape_index_001")
  17. # 创建索引并定义 Mapping
  18. mapping = {
  19.     "mappings": {
  20.         "properties": {
  21.             "location": {
  22.                 "type": "geo_shape"
  23.             },
  24.             "centroid": {  # 新增字段
  25.                 "type": "geo_point"
  26.             },
  27.             "name": {
  28.                 "type": "text"
  29.             }
  30.         }
  31.     }
  32. }
  34. # 创建索引
  35. es.indices.create(index="province2023_centroid_geoshape_index_001", body=mapping)
  37. # 读取 GeoJSON 文件
  38. with open("D:/data/gis/2023-province-GeoJSON.gesjson", "r", encoding="utf-8") as file:
  39.     geojson_data = json.load(file)
  41. # 提取 GeoJSON 特征集合
  42. features = geojson_data.get("features", [])
  45. def calculate_centroid(polygons):
  46.     total_area = 0.0
  47.     total_x = 0.0
  48.     total_y = 0.0
  49.  
  50.     for polygon in polygons:
  51.         # 现在 polygon 是一个包含多个坐标的列表
  52.         centroid = calculate_simple_polygon_centroid(polygon)
  53.         area = calculate_polygon_area(polygon)
  55.         total_area += area
  56.         total_x += centroid[0] * area
  57.         total_y += centroid[1] * area
  59.     if total_area == 0:
  60.         # 如果总面积为零，则返回原点作为中心点
  61.         return [0, 0]
  62.     else:
  63.         return [total_x / total_area, total_y / total_area]
  66. # is_coordinates_list方法
  67. # 以下结构返回True，polygon 是一个包含坐标列表的列表
  68. # [
  69. #     [[x1, y1], [x2, y2], [x3, y3], ...],
  70. #     [[x1, y1], [x2, y2], [x3, y3], ...]  # 如果有内部孔洞
  71. # ]
  72. # 以下结构返回Fasle，包含单个坐标的列表
  73. # [
  74. #     [x1, y1],
  75. #     [x2, y2],
  76. #     [x3, y3],
  77. #     ...
  78. # ]
  80. def is_coordinate(coord):
  81.     return (
  82.             isinstance(coord, (list, tuple)) and
  83.             len(coord) == 2 and
  84.             all(isinstance(c, (int, float)) for c in coord)
  85.     )
  88. def is_coordinates_list(coords):
  89.     # 检查 coords 是否是一个包含坐标列表的列表
  90.     if isinstance(coords, list):
  91.         if all(isinstance(c, list) and all(is_coordinate(coord) for coord in c) for c in coords):
  92.             return True
  93.     return False
  96. def calculate_simple_polygon_centroid(polygon):
  97.     # 确定 polygon 的结构
  98.     if is_coordinates_list(polygon):
  99.         # polygon 是一个包含坐标列表的列表
  100.         x_sum = sum(coord[0] for coord in polygon[0])
  101.         y_sum = sum(coord[1] for coord in polygon[0])
  102.         num_points = len(polygon[0])
  103.     else:
  104.         # print(False, polygon[0])
  105.         # polygon 是一个包含多个坐标的列表
  106.         x_sum = sum(coord[0] for coord in polygon)
  107.         y_sum = sum(coord[1] for coord in polygon)
  108.         num_points = len(polygon)
  109.     # 计算平均坐标
  110.     centroid_x = x_sum / num_points
  111.     centroid_y = y_sum / num_points
  113.     return [centroid_x, centroid_y]
  116. def calculate_polygon_area(polygon):
  117.     # 计算简单多边形的面积
  118.     area = 0.0
  119.     if is_coordinates_list(polygon):  # polygon 是一个包含坐标列表的列表
  120.         num_points = len(polygon[0])
  121.         for i in range(num_points):
  122.             j = (i + 1) % num_points
  123.             area += polygon[0][i][0] * polygon[0][j][1]
  124.             area -= polygon[0][j][0] * polygon[0][i][1]
  126.     else:  # polygon 是一个包含多个坐标的列表
  127.         num_points = len(polygon)
  128.         for i in range(num_points):
  129.             j = (i + 1) % num_points
  130.             area += polygon[i][0] * polygon[j][1]
  131.             area -= polygon[j][0] * polygon[i][1]
  132.     return abs(area) / 2.0
  135. # 准备数据以供导入
  136. documents = []
  137. for feature in features:
  138.     # 检查坐标是否在有效范围内
  139.     coordinates = feature["geometry"]["coordinates"]
  140.     centroid = calculate_centroid(coordinates)
  142.     doc = {
  143.         "location": {
  144.             "type": feature["geometry"]["type"],
  145.             "coordinates": coordinates
  146.         },
  147.         "centroid": centroid,  # 添加中心点
  148.     }
  149.     if "properties" in feature:
  150.         doc.update(feature["properties"])
  151.     documents.append(doc)
  153. # 定义批量大小
  154. batch_size = 100  # 每次批量导入的数量
  157. # 准备 actions
  158. def generate_actions(documents):
  159.     for doc in documents:
  160.         yield {
  161.             "_index": "district2023_centroid_geoshape_index_001",
  162.             "_source": doc
  163.         }
  166. # 分批执行批量导入
  167. for i in range(0, len(documents), batch_size):
  168.     end = min(i + batch_size, len(documents))
  169.     success, errors = bulk(es, generate_actions(documents[i:end]))
  170.     if errors:
  171.         print(f"Bulk {i}-{end} completed, {success} documents indexed, but {len(errors)} documents failed.")
  172.         for error in errors:
  173.             print(error)
  174.     else:
  175.         print(f"Bulk {i}-{end} completed, {success} documents indexed.")
  177. print("All data indexed.")

第五步：利用elasticsearch的pipeline和reindex能力预处理数据

  1. # geo_centroid 聚合是一种高级聚合，它可以计算一组地理位置的中心点。在 Elasticsearch 中，这个功能属于高级特性，通常只在 X-Pack（现在称为 Elastic Security 和 Elastic Observability）的许可证中可用。
  2. # 试用30天可以体验
  3. POST /province2023_geoshape_index_001/_search
  4. {
  5.   "size": 0,
  6.   "aggs": {
  7.     "centroid": {
  8.       "geo_centroid": {
  9.         "field": "location"
  10.       }
  11.     }
  12.   }
  13. }
  16. POST province2023_centroid_geoshape_index_001/_search
  17. {
  18.   "query": {
  19.     "term": {
  20.       "省.keyword": {
  21.         "value": "陕西省" 
  22.       }
  23.     }
  24.   }
  25. }
  27. PUT _ingest/pipeline/copy_field_pipeline
  28. {
  29.   "description": "Copy the value of one field to another",
  30.   "processors": [
  31.     {
  32.       "copy": {
  33.         "from": "省", 
  34.         "to": "province_name"
  35.       }
  36.     }
  37.   ]
  38. }
  39. GET province2023_centroid_geoshape_index_001/_mapping
  41. GET province2023_centroid_geoshape_index_001/_mapping
  44. PUT _ingest/pipeline/province_multiple_copy_fields_pipeline
  45. {
  46.   "description": "Copy multiple fields to new fields and rename fields to new fields",
  47.   "processors": [
  48.     {
  49.       "set": {
  50.         "field": "province_name",
  51.         "value": "{{{省}}}"
  52.       }
  53.     },
  54.     {
  55.       "remove": {
  56.         "field": "省"
  57.       }
  58.     },
  59.     {
  60.       "rename": {
  61.         "field": "省级码",
  62.         "target_field": "province_code"
  63.       }
  64.     },
  65.     {
  66.       "rename": {
  67.         "field": "省类型",
  68.         "target_field": "province_type"
  69.       }
  70.     },
  71.     {
  72.       "rename": {
  73.         "field": "VAR_NAME",
  74.         "target_field": "var_name"
  75.       }
  76.     },
  77.     {
  78.       "rename": {
  79.         "field": "ENG_NAME",
  80.         "target_field": "eng_name"
  81.       }
  82.     },
  83.     {
  84.       "rename": {
  85.         "field": "FIRST_GID",
  86.         "target_field": "first_gid"
  87.       }
  88.     },
  89.     {
  90.       "rename": {
  91.         "field": "FIRST_TYPE",
  92.         "target_field": "first_type"
  93.       }
  94.     }
  95.   ]
  96. }
  98. GET province2023_centroid_geoshape_index_002/_count
  100. GET province2023_centroid_geoshape_index_002/_mapping
  101. DELETE province2023_centroid_geoshape_index_002
  103. PUT province2023_centroid_geoshape_index_002
  104. {
  105.   "mappings": {
  106.     "properties": {
  107.       "eng_name": {
  108.         "type": "text",
  109.         "fields": {
  110.           "keyword": {
  111.             "type": "keyword",
  112.             "ignore_above": 256
  113.           }
  114.         }
  115.       },
  116.       "first_gid": {
  117.         "type": "text",
  118.         "fields": {
  119.           "keyword": {
  120.             "type": "keyword",
  121.             "ignore_above": 256
  122.           }
  123.         }
  124.       },
  125.       "first_type": {
  126.         "type": "text",
  127.         "fields": {
  128.           "keyword": {
  129.             "type": "keyword",
  130.             "ignore_above": 256
  131.           }
  132.         }
  133.       },
  134.       "var_name": {
  135.         "type": "text",
  136.         "fields": {
  137.           "keyword": {
  138.             "type": "keyword",
  139.             "ignore_above": 256
  140.           }
  141.         }
  142.       },
  143.       "centroid": {
  144.         "type": "geo_point"
  145.       },
  146.       "location": {
  147.         "type": "geo_shape"
  148.       },
  149.       "name": {
  150.         "type": "text"
  151.       },
  152.       "year": {
  153.         "type": "text",
  154.         "fields": {
  155.           "keyword": {
  156.             "type": "keyword",
  157.             "ignore_above": 256
  158.           }
  159.         }
  160.       }
  161.     }
  162.   }
  163. }
  165. POST _reindex
  166. {
  167.   "source": {
  168.     "index": "province2023_centroid_geoshape_index_001"
  169.   },
  170.   "dest": {
  171.     "index": "province2023_centroid_geoshape_index_002",
  172.     "pipeline": "province_multiple_copy_fields_pipeline"
  173.   }
  174. }
  176. GET province2023_centroid_geoshape_index_002/_search

第六步：查询数据 geo_distance

  1. # centroid字段的type是 geo_point，存储的经纬度形式是数组Geopoint as an array
  2. # geo_bounding_box 可查找边框内的所有地理坐标点。
  3. POST province2023_centroid_geoshape_index_002/_search
  4. {
  5.   "query": {
  6.     "geo_bounding_box": { 
  7.       "centroid": {
  8.         "top_left": {
  9.           "lat": 42,
  10.           "lon": -72
  11.         },
  12.         "bottom_right": {
  13.           "lat": 40,
  14.           "lon": -74
  15.         }
  16.       }
  17.     }
  18.   }
  19. }
  21. POST province2023_centroid_geoshape_index_002/_search
  22. {
  23.   "query": {
  24.     "geo_distance": {
  25.       "distance": 100,
  26.       "centroid": {
  27.         "lat": 40.09937484066758,
  28.         "lon": 116.41960604340115
  29.       }
  30.     }
  31.   }
  32. }
  34. POST province2023_centroid_geoshape_index_002/_search
  35. {
  36.   "query": {
  37.     "bool": {
  38.       "must": {
  39.         "match": {
  40.           "province_name":"xx市"
  41.         }
  42.       },
  43.       "filter": {
  44.         "geo_distance": {
  45.           "distance": "2km",
  46.           "centroid": {
  47.             "lat": 40.09937484066758,
  48.             "lon": 116.41960604340115
  49.           }
  50.         }
  51.       }
  52.     }
  53.   }
  54. }
  57. POST province2023_centroid_geoshape_index_002/_search
  58. {
  59.   "query": {
  60.     "bool": {
  61.       "must": {
  62.         "match": {
  63.           "province_name":"xx市"
  64.         }
  65.       },
  66.       "filter": {
  67.         "geo_distance": {
  68.           "distance": "200km",
  69.           "location": {
  70.             "lat": 40.09937484066758,
  71.             "lon": 116.41960604340115
  72.           }
  73.         }
  74.       }
  75.     }
  76.   }
  77. }

 

原文链接：https://www.cnblogs.com/hbuuid/p/18327500

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Elasticsearch热门文章

  • ES 14 - (底层原理) Elasticsearch内部如何处理不同type的数据
  • ES 15 - Elasticsearch中的数据类型 (text、keyword、date、geo等)
  • java中elasticsearch7.x关于nested类型的api使用，新增+更新
  • Elasticsearch 映射参数详解 fields
  • [Elasticsearch] ES聚合场景下部分结果数据未返回问题分析
  • ES 22 - Elasticsearch中如何进行日期(数值)范围查询
  • Elasticsearch查询之Term?Query示例解析
  • ElasticSearch 实现分词全文检索 - SpringBoot 完整实现 Demo 附源码【完结篇】
  • ElasticStack学习（十）：深入ElasticSearch搜索之QueryFiltering、多/单字符串的多字段查询
  • ElasticSearch7 设置外网访问

    • Elasticsearch推荐文章

  • 利用Elasticsearch实现地理位置、城市搜索服务
  • 日志采集/分析
  • ELKF（elasticsearch、logstash、kibana、filebeat）搭建及收集nginx日志特昂糖-
  • ElasticSearch性能原理拆解
  • Elasticsearch使用实战以及代码详解
  • 使用 Docker Compose V2 快速搭建日志分析平台 ELK (Elasticsearch、Logstash 和 Kibana)
  • 【解决方案】MySQL5.7 百万数据迁移到 ElasticSearch7.x 的思考
  • K8s部署轻量级日志收集系统EFK（elasticsearch + filebeat + kibana）
  • Docker下elasticsearch8部署、扩容、基本操作实战(含kibana)
  • Elasticsearch 保姆级入门篇

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号