Elasticsearch(GEO)数据写入和空间检索

Elasticsearch简介

什么是 Elasticsearch?
Elasticsearch 是一个开源的分布式 RESTful搜索和分析引擎，能够解决越来越多不同的应用场景。

本文内容

本文主要是介绍了ES GEO数据写入和空间检索，ES版本为7.3.1

数据准备

Qgis使用渔网工具，对范围进行切割，得到网格的Geojson

新建索引设置映射

def set_mapping(es,index_name="content_engine",doc_type_name="en",my_mapping={}):
    # ignore 404 and 400
    es.indices.delete(index=index_name, ignore=[400, 404])
    print("delete_index")
    # ignore 400 cause by IndexAlreadyExistsException when creating an index
    my_mapping = {
        "properties": {
            "location": {"type": "geo_shape"},
            "id": {"type": "long"}
        }
    }
    create_index = es.indices.create(index=index_name)
    mapping_index = es.indices.put_mapping(index=index_name, doc_type=doc_type_name, body=my_mapping,                          include_type_name=True)
    print("create_index")
    if create_index["acknowledged"] is not True or mapping_index["acknowledged"] is not True:
        print("Index creation failed...")

数据插入

使用multiprocessing和elasticsearch.helpers.bulk进行数据写入,每一万条为一组写入，剩下的为一组,然后多线程写入。分别写入4731254条点和面数据。写入时候使用多核，ssd，合适的批量数据可以有效加快写入速度，通过这些手段可以在三分钟左右写入四百多万的点或者面数据。

def mp_worker(features):
    count = 0
    es = Elasticsearch(hosts=[ip], timeout=5000)
    success, _ = bulk(es,features, index=index_name, raise_on_error=True)
    count += success
    return count
def mp_handler(input_file, index_name, doc_type_name="en"):
    with open(input_file, 'rb') as f:
        data = json.load(f)
    features = data["features"]
    del data
    act=[]
    i=0
    count=0
    actions = []
    for feature in features:
        action = {
                "_index": index_name,
                "_type": doc_type_name,
                "_source": {
                    "id": feature["properties"]["id"],
                    "location": {
                        "type": "polygon",
                        "coordinates": feature["geometry"]["coordinates"]
                    }
                }
            }
        i=i+1
        actions.append(action)
        if (i == 9500):
            act.append(actions)
            count=count+i
            i = 0
            actions = []
    if i!=0:
        act.append(actions)
        count = count + i
    del features
    print('read all %s data ' % count)
    p = multiprocessing.Pool(4)
    i=0
    for result in p.imap(mp_worker, act):
        i=i+result
    print('write all %s data ' % i)

GEO（point）查询距离nkm附近的点和范围选择

from elasticsearch import Elasticsearch
from elasticsearch.helpers import scan
import time
starttime = time.time()
_index = "gis_point"
_doc_type = "20190824"
ip = "127.0.0.1:9200"
# 附近nkm 选择
_body = {
    "query": {
        "bool": {
            "must": {
                "match_all": {}
            },
            "filter": {
                "geo_distance": {
                    "distance": "9km",
                    "location": {
                        "lat": 18.1098857850465471,
                        "lon": 109.1271036098896730
                    }
                }
            }
        }
    }
}
# 范围选择
# _body={
#   "query": {
#     "geo_bounding_box": {
#       "location": {
#         "top_left": {
#           "lat": 18.4748659238899933,
#           "lon": 109.0007435371629470
#         },
#         "bottom_right": {
#           "lat": 18.1098857850465471,
#           "lon": 105.1271036098896730
#         }
#       }
#     }
#   }
# }
es = Elasticsearch(hosts=[ip], timeout=5000)
scanResp = scan(es, query=_body, scroll="10m", index=_index, timeout="10m")
for resp in scanResp:
    print(resp)
endtime = time.time()
print(endtime - starttime)

GEO（shape）范围选择

from elasticsearch import Elasticsearch
from elasticsearch.helpers import scan
import time
starttime = time.time()
_index = "gis"
_doc_type = "20190823"
ip = "127.0.0.1:9200"
# envelope format, [[minlon,maxlat],[maxlon,minlat]]
_body = {
    "query": {
        "bool": {
            "must": {
                "match_all": {}
            },
            "filter": {
                "geo_shape": {
                    "location": {
                        "shape": {
                            "type": "envelope",
                            "coordinates": [[108.987103609889, 18.474865923889993], [109.003537162947, 18.40988578504]]
                        },
                        "relation": "within"
                    }
                }
            }
        }
    }
}

es = Elasticsearch(hosts=[ip], timeout=5000)
scanResp = scan(es, query=_body, scroll="1m", index=_index, timeout="1m")
for resp in scanResp:
    print(resp)
endtime = time.time()
print(endtime - starttime)

GEO（point）距离聚合

from elasticsearch import Elasticsearch
import time
starttime = time.time()
_index = "gis_point"
_doc_type = "20190824"
ip = "127.0.0.1:9200"
# 距离聚合
_body = {
    "aggs" : {
        "rings_around_amsterdam" : {
            "geo_distance" : {
                "field" : "location",
                "origin" : "18.1098857850465471,109.1271036098896730",
                "ranges" : [
                    { "to" : 100000 },
                    { "from" : 100000, "to" : 300000 },
                    { "from" : 300000 }
                ]
            }
        }
    }
}

es = Elasticsearch(hosts=[ip], timeout=5000)
scanResp = es.search( body=_body, index=_index)
for i in scanResp['aggregations']['rings_around_amsterdam']['buckets']:
    print(i)
endtime = time.time()
print(endtime - starttime)

中心点聚合

_body ={
     "aggs" : {
        "centroid" : {
            "geo_centroid" : {
                "field" : "location"
            }
        }
    }
}
es = Elasticsearch(hosts=[ip], timeout=5000)
scanResp = es.search( body=_body, index=_index)
print(scanResp['aggregations'])

范围聚合

_body = {
    "aggs": {
        "viewport": {
            "geo_bounds": {
                "field": "location"

            }
        }
    }
}
es = Elasticsearch(hosts=[ip], timeout=5000)
scanResp = es.search(body=_body, index=_index)
print(scanResp['aggregations']['viewport'])

geohash聚合

##低精度聚合，precision代表geohash长度
_body = {
    "aggregations": {
        "large-grid": {
            "geohash_grid": {
                "field": "location",
                "precision": 3
            }
        }
    }
}
# 高精度聚合,范围聚合以及geohash聚合
# _body = {
#     "aggregations": {
#         "zoomed-in": {
#             "filter": {
#                 "geo_bounding_box": {
#                     "location": {
#                         "top_left": "18.4748659238899933,109.0007435371629470",
#                         "bottom_right": "18.4698857850465471,108.9971036098896730"
#                     }
#                 }
#             },
#             "aggregations": {
#                 "zoom1": {
#                     "geohash_grid": {
#                         "field": "location",
#                         "precision": 7
#                     }
#                 }
#             }
#         }
#     }
# }
es = Elasticsearch(hosts=[ip], timeout=5000)
scanResp = es.search(body=_body, index=_index)
for i in scanResp['aggregations']['large-grid']['buckets']:
    print(i)
#for i in scanResp['aggregations']['zoomed-in']['zoom1']['buckets']:
#    print(i)    

切片聚合

# 低精度切片聚合,precision代表级别
_body = {
    "aggregations": {
        "large-grid": {
            "geotile_grid": {
                "field": "location",
                "precision": 8
            }
        }
    }
}
# 高精度切片聚合，范围聚合以切片聚合
# _body={
#     "aggregations" : {
#         "zoomed-in" : {
#             "filter" : {
#                 "geo_bounding_box" : {
#                     "location" : {
#                         "top_left": "18.4748659238899933,109.0007435371629470",
#                          "bottom_right": "18.4698857850465471,108.9991036098896730"
#                     }
#                 }
#             },
#             "aggregations":{
#                 "zoom1":{
#                     "geotile_grid" : {
#                         "field": "location",
#                         "precision": 18
#                     }
#                 }
#             }
#         }
#     }
# }
es = Elasticsearch(hosts=[ip], timeout=5000)
scanResp = es.search(body=_body, index=_index)
for i in scanResp['aggregations']['large-grid']['buckets']:
    print(i)
# for i in scanResp['aggregations']['zoomed-in']['zoom1']['buckets']:
#      print(i)

Elasticsearch和PostGIS相同功能对比

PostGIS最近点查询

SELECT  id,geom, ST_DistanceSphere(geom,'SRID=4326;POINT(109.1681036098896730 18.1299957850465471)'::geometry) 
FROM  h5 
ORDER BY  geom <->
'SRID=4326;POINT(109.1681036098896730 18.1299957850465471)'::geometry
LIMIT 1 

Elasticsearch最近点查询

from elasticsearch import Elasticsearch
import time
starttime = time.time()
_index = "gis_point"
_doc_type = "20190824"
ip = "127.0.0.1:9200"

_body={
  "sort": [
    {
      "_geo_distance": {
        "unit": "m",
        "order": "asc",
        "location": [
          109.1681036098896730,
          18.1299957850465471
        ],
        "distance_type": "arc",
        "mode": "min",
        "ignore_unmapped": True
      }
    }
  ],
  "from": 0,
  "size": 1,
    "query": {
        "bool": {
          "must": {
            "match_all": {}
          }
        }
      }

}
es = Elasticsearch(hosts=[ip], timeout=5000)
scanResp = es.search(body=_body, index=_index)
endtime = time.time()
print(endtime - starttime)

PostGIS范围查询

select id,geom,fid  FROM public."California"
where 
ST_Intersects(geom,ST_MakeEnvelope(-117.987103609889,33.40988578504,-117.003537162947,33.494865923889993, 4326))=true
[-117.987103609889, 33.494865923889993], [-117.003537162947, 33.40988578504]

Elasticsearch范围查询

from elasticsearch import Elasticsearch
from elasticsearch.helpers import scan
import time
starttime = time.time()
_index = "gis_california"
ip = "127.0.0.1:9200"
# envelope format, [[minlon,maxlat],[maxlon,minlat]]

_body = {
    "query": {
        "bool": {
            "must": {
                "match_all": {}
            },
            "filter": {
                "geo_shape": {
                    "geom": {
                        "shape": {
                            "type": "envelope",
                            "coordinates": [[-117.987103609889, 33.494865923889993], [-117.003537162947, 33.40988578504]]
                        },
                        "relation": "INTERSECTS"
                    }
                }
            }
        }
    }
}
es = Elasticsearch(hosts=[ip], timeout=5000)
scanResp = scan(es, query=_body, scroll="1m", index=_index, timeout="1m")
i=0
for resp in scanResp:
    i=i+1
    a=resp
print(i)
endtime = time.time()
print(endtime - starttime)

两种场景中PostGIS的性能更好


参考资料：

1.Elasticsearch(GEO)空间检索查询

2.Elasticsearch官网

3.PostGIS拆分LineString为segment,point

4.亿级“附近的人”，打通“特殊服务”通道

5.PostGIS教程二十二：最近邻域搜索