Code
import geojson
from datetime import datetime
from elasticsearch import Elasticsearch, helpers
def geojson_to_es(gj):
for feature in gj['features']:
date = datetime.strptime("-".join(feature["properties"]["event_date"].split('-')[0:2]) + "-" + feature["properties"]["year"], "%d-%b-%Y")
feature["properties"]["timestamp"] = int(date.timestamp())
feature["properties"]["event_date"] = date.strftime('%Y-%m-%d')
yield feature
with open("GeoObs.json") as f:
gj = geojson.load(f)
es = Elasticsearch(hosts=[{'host': 'localhost', 'port': 9200}])
k = ({
"_index": "YOUR_INDEX",
"_source": feature,
} for feature in geojson_to_es(gj))
helpers.bulk(es, k)
解释
with open("GeoObs.json") as f:
gj = geojson.load(f)
es = Elasticsearch(hosts=[{'host': 'localhost', 'port': 9200}])
这部分代码加载外部 geojson 文件，然后连接到 Elasticsearch。
k = ({
"_index": "conflict-data",
"_source": feature,
} for feature in geojson_to_es(gj))
helpers.bulk(es, k)
The ()s 这里创建了一个我们将提供给它的生成器helpers.bulk(es, k)。记住_source是 Elasticsearch 中的原始数据 - IE：我们的原始 JSON。_index只是我们想要放置数据的索引。您将看到其他示例_doc这里。这是映射类型的一部分，在 Elasticsearch 7.X+ 中不再存在。
def geojson_to_es(gj):
for feature in gj['features']:
date = datetime.strptime("-".join(feature["properties"]["event_date"].split('-')[0:2]) + "-" + feature["properties"]["year"], "%d-%b-%Y")
feature["properties"]["timestamp"] = int(date.timestamp())
feature["properties"]["event_date"] = date.strftime('%Y-%m-%d')
yield feature
功能geojson使用生成器来生成事件。生成器函数将代替返回和完成resume at the keyword每次调用后yield`。在本例中，我们正在生成 GeoJSON 特征。在我的代码中您还可以看到：
date = datetime.strptime("-".join(feature["properties"]["event_date"].split('-')[0:2]) + "-" + feature["properties"]["year"], "%d-%b-%Y")
feature["properties"]["timestamp"] = int(date.timestamp())
feature["properties"]["event_date"] = date.strftime('%Y-%m-%d')
这只是在将 JSON 数据发送到 Elasticsearch 之前对其进行操作的示例。
关键在你的映射文件中你必须有一些标记为geo_point or geo_shape。这些数据类型是 Elasticsearch 识别地理数据的方式。我的映射文件中的示例：
...
{
"properties": {
"geometry": {
"properties": {
"coordinates": {
"type": "geo_point"
},
"type": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
}
}
},
...
也就是说，在使用 Python 上传 GeoJSON 数据之前，您需要创建索引，然后应用一个映射文件，其中包括geo_shape or geo_point使用类似的东西：
curl -X PUT "localhost:9200/YOUR_INDEX?pretty"
curl -X PUT localhost:9200/YOUR_INDEX/_mapping?pretty -H "Content-Type: application/json" -d @mapping.json
