Elasticsearch(GEO)數據寫入和空間檢索

来源:https://www.cnblogs.com/polong/archive/2019/09/15/11523955.html
-Advertisement-
Play Games

Elasticsearch簡介 什麼是 Elasticsearch? Elasticsearch 是一個開源的分散式 RESTful搜索和分析引擎，能夠解決越來越多不同的應用場景。 本文內容 本文主要是介紹了ES GEO數據寫入和空間檢索，ES版本為7.3.1 數據準備 Qgis使用漁網工具，對範圍 ...


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教程二十二：最近鄰域搜索


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 2019-2020-1 20199319《Linux內核原理與分析》第一周作業
    一、Linux系統簡介 通過實驗一瞭解了Linux 的歷史，Linux與windows之間的區別以及學習Linux的方法。因為一直用的都是windows系統，習慣了圖形界面，而Linux是通過輸入命令執行操作，所以初學還很不適應。正如那句話說的windows能做的Linux都能做，windows不能 ...
  • 記錄舊手機(米6)安裝Linux(Ubuntu16.04)踩過的坑
    舊的小米6在抽屜吃灰半年，一直沒想好要怎麼處理，於是就想著安裝Linux。 完整教程來自https://blog.csdn.net/Greepex/article/details/85333027 原文里把每一個步驟都描述得很清楚（所以本文就不貼詳細步驟圖了，豎版截圖太影響觀感），但難免會踩一些坑。 ...
  • LINUX基礎操作
    常用的linux命令 ls 查看當前（或者指定）目錄下的文件列表 ls -l 查看詳細信息列表 ls -a 或ls -al 查看目錄下所有文件（包含隱藏文件）的詳細信息 cd ./ 切換到當前目錄 cd ../ 切換到上一級目錄 clear 清屏 （或者ctrl+l） / 根目錄 ~ 家目錄 cd ...
  • windows控制台，cmd，命令提示符下的基礎操作
    打開dos命令視窗1、win+r-->運行-->cmd 2、摁住shift+滑鼠右擊 選擇 在此處打開命令視窗3、在磁碟某文件夾下，選擇標題欄中輸入框，輸入cmd 回車 windows下常用的命令 系統管理和文件管理systeminfo 獲取系統信息 系統 補丁 網卡path 查看環境變數set 查 ...
  • VM虛擬機下centos安裝。
    centOS 7安裝步驟： 1.選擇新建虛擬機，稍後安裝，linux選centos7 64位 2.位置改到存放虛擬機的文件夾 3.把硬碟空間改到40g，記憶體分到4g，1處理器2個核心 4 更改cd/dvd到鏡像位置。 5 選擇中文安裝 6選擇需要的安裝軟體，gui和gnome桌面 7 設置root密 ...
  • 【原創】（四）Linux記憶體模型之Sparse Memory Model
    背景 By 魯迅 By 高爾基 說明： 1. Kernel版本：4.14 2. ARM64處理器，Contex A53，雙核 3. 使用工具：Source Insight 3.5， Visio 1. 介紹 順著之前的分析，我們來到了 函數了，本以為一篇文章能搞定，大概掃了一遍代碼之後，我默默的把它拆 ...
  • 生產者-消費者問題
    [TOC] 1. 概述 定義 生產者消費者問題是線程同步的經典問題，也稱為有界緩衝區問題，問題描述大致如下： 生產者和消費者之間共用一個有界數據緩衝區 一個或多個生產者（線程或進程）向緩衝區放置數據 一個或多個消費者（線程或進程）從緩衝區取出數據 緩衝區 生產者消費者問題中的緩衝區，包括隊列緩衝區和 ...
  • Posix信號量
    [TOC] 1. Posix IPC 概述 以下三種類型的IPC合稱為Posix IPC： Posix信號量 Posix消息隊列 Posix共用記憶體 Posix IPC在訪問它們的函數和描述它們的信息上有一些類似點，主要包括： IPC名字 創建或打開時指定的讀寫許可權、創建標誌以及用戶訪問許可權 下表匯 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類