导入json数据到es
- 每条记录必须有一条对应的index声明。json文件应符合Bulk API要求
- json文件中每条数据前需包含index信息（index/type/id...）
curl -H 'Content-Type: application/x-ndjson' -s -XPOST localhost:9200/_bulk --data-binary @accounts.json
{"index":{"_index":"index2","_type":"type2","_id":0}}
{"age":10,"name":"jim"}
{"index":{"_index":"index2","_type":"type2","_id":1}}
{"age":16,"name":"tom"}
- 若导入文件中为同一个index（/index/type），则可在url中声明默认index（/index/type），则json文件中无须声明index和type，同时，支持部分记录显示声明index和type。
curl -H 'Content-Type: application/x-ndjson' -s -XPOST localhost:9200/index1/_bulk --data-binary @accounts.json #URL中声明默认index
curl -H 'Content-Type: application/x-ndjson' -s -XPOST localhost:9200/index1/type1/_bulk --data-binary @accounts.json #URL中声明默认index/type
{"index":{}
{"age":6,"name":"bob"}
{"index":{"_id":"2"}}
{"age":10,"name":"jim"}
{"index":{"_id":"6"}}
{"index":{"_type":"type2","_id":1}}
{"age":16,"name":"tom"}
{"index":{"_index":"index2","_type":"type3","_id":1}}
{"age":20,"name":"lucy"}
- 可直接导入符合格式要求的json，无须事先创建mapping，es可根据json数据自动创建。
也可事先声明mapping，，进行自定义设置 - curl命令导入示例
导入shapefile到es
查阅众多资料后，理论上有如下几种方式：
- 使用GDAL直接将shapefile导入ES，失败
- 使用GDAL（或ArcMap）将shapefile导成json文件，使用curl命令导入json文件
说明：生成的json文件格式不符合Bulk API要求，需处理后方能导入 - 使用Arcpy编写脚本，将shapefile导出成符合Bulk要求的json，再使用curl命令导入（或脚本实现）
- 使用ArcMap导出geojson（或使用Arcpy），使用python（java）解析json文件，使用Bulk API编写脚本导入ES
其中需注意的问题为：
- 使用GDAL或ArcMap导出的geojson格式和Bulk API要求的数据格式不同
- 数据需符合Geo-shape datatype或Geo-point datatype方能使用curl命令导入
综上，最终选择使用第4种方法解决，使用GDAL将shapefile导出成geojson文件，再使用python elasticsearch bulk API编写脚本，解析geojson并导入ES。
1. shapefile to es（失败）
GDAL for ES Driver提供shapefile直接导入ES的方法，但导入时报错
ogr2ogr -f "ElasticSearch" http://localhost:9200 my_shapefile.shp
#ERROR 1: HTTP error code : 405
#ERROR 8: Could not connect to server
#ElasticSearch driver failed to create http://localhost:9200/
2. 通过json文件中转（需处理json文件格式）
另外一种方式为使用ogr2ogr工具将shapefile转换为geojson文件，再将geojson文件导入ES。
2.1 shapefile to geojson
How to convert and import Arc Shapefile from Zillow into an elastic search database?
Ask
ogr2ogr -f GeoJSON map1.geojson map.shp
生成的json文件格式如下，不包含index信息，Bulk API无法直接导入。
{
"type": "FeatureCollection",
"features": [
{ "type": "Feature", "properties": { "ID_0": 45, "NAME_1": "Anhui" }, "geometry": { "type": "Polygon", "coordinates": [ [ [ 119.632111, 31.139344000000108 ], [ 119.644439000000148, 31.115657 ], [ 119.624672, 31.084624000000133 ] ] ] } }
{ "type": "Feature", "properties": { "ID_0": 45, "NAME_1": "Beijing" }, "geometry": { "type": "Polygon", "coordinates": [ [ [ 117.379737, 40.226871 ], [ 117.382896, 40.208718000000147 ], [ 117.369484, 40.190997 ] ] ] } }
]
}
2.2 create index mapping
Indexing Geo Shapes
可针对geo字段和重点字段进行mapping，其余字段导入时自动生成。
PUT /gis1
{
"mappings": {
"province1": {
"properties": {
"location": {
"type": "geo_shape"
},
"ID_0": {
"type": "text"
},
"NAME_1": {
"type": "text"
}
}
}
}
}
2.3 import json
curl -H 'Content-Type: application/x-ndjson' -XPOST 'http://localhost:9200/gis1/province1/_bulk?pretty' --data-binary @map1.geojson
#报错："type":"json_e_o_f_exception","reason":"Unexpected end-of-input: expected close marker for Object (start marker at [Source: org.elasticsearch.transport.netty4.ByteBufStreamInput@1ffa375d; line: 1, column: 1])\n at [Source: org.elasticsearch.transport.netty4.ByteBufStreamInput@1ffa375d; line: 1, column: 3]"
需将数据改造成如下格式后，再使用bulk命令导入。
{"index":{"_id":"1"}}
{ "type": "Feature", "properties": { "ID_0": 45, "NAME_1": "Zhejiang" }, "geometry": { "type": "MultiPolygon", "coordinates": [ [ [ [ 121.187362, 27.86903100000012 ], [ 121.190140000000156, 27.847919 ], [ 121.156249, 27.823749000000134 ]] ] ] } }
{"index":{"_id":"2"}}
{ "type": "Feature", "properties": { "ID_0": 46, "NAME_1": "Jiangsu" }, "geometry": { "type": "MultiPolygon", "coordinates": [ [ [ [ 121.187362, 27.86903100000012 ], [ 121.190140000000156, 27.847919 ], [ 121.156249, 27.823749000000134 ]] ] ] } }
4. python脚本导入GDAL生成的geojson文件
# rasa 20180212
# 使用ES python api插入geojson面数据（map.geojson）
import json
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk
def set_mapping(es, index_name="content_engine", doc_type_name="en"):
my_mapping = {
doc_type_name: {
"properties": {
"location": {
"type": "geo_shape"
},
"ID_0": {
"type": "text"
},
"NAME_1": {
"type": "text"
}
}
}
}
# 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
create_index = es.indices.create(index=index_name, ignore=400)
mapping_index = es.indices.put_mapping(index=index_name, doc_type=doc_type_name, body=my_mapping)
if create_index["acknowledged"] is not True or mapping_index["acknowledged"] is not True:
print("Index creation failed...")
def set_data(es, input_file, index_name, doc_type_name="en"):
with open(input_file, 'r') as f:
data = json.load(f)
features = data["features"]
ACTIONS = []
i = 0
count = 0
for feature in features:
action = {}
if (feature["geometry"]["type"] == "Polygon"): # 判断geometry类型为polygon
action = {
"_index": index_name,
"_type": doc_type_name,
"_source": {
"ID_0": feature["properties"]["ID_0"],
"NAME_1": feature["properties"]["NAME_1"],
"location": {
"type": "polygon",
"coordinates": feature["geometry"]["coordinates"]
}
}
}
else: # geometry类型为multipolygon
action = {
"_index": index_name,
"_type": doc_type_name,
"_source": {
"ID_0": feature["properties"]["ID_0"],
"NAME_1": feature["properties"]["NAME_1"],
"location": {
"type": "multipolygon",
"coordinates": feature["geometry"]["coordinates"]
}
}
}
i += 1
print("prepare insert: %s" % feature["properties"]["NAME_1"])
print("type: %s" % feature["geometry"]["type"])
ACTIONS.append(action)
if (i == 5):
success, _ = bulk(es, ACTIONS, index=index_name, raise_on_error=True)
count += success
i = 0
ACTIONS = []
print("insert %s lines" % count)
success, _ = bulk(es, ACTIONS, index=index_name, raise_on_error=True)
count += success
print("insert %s lines" % count)
if __name__ == '__main__':
# es = Elasticsearch(hosts=["127.0.0.1:9200"], http_auth=('elastic','changeme'),timeout=5000)
es = Elasticsearch(hosts=["127.0.0.1:9200"], timeout=5000)
set_mapping(es, "gis6", "province")
# geojson文件为ogr2ogr生成格式
# set_data(es, "./data/map-fujian.geojson", "gis6", "province") # multipolygon
# set_data(es, "./data/map-anhui.geojson", "gis6", "province") # polygon
set_data(es, "./data/map-full.geojson", "gis6", "province") # polygon & multipolygon
# set_data(es, "./data/map.geojson", "gis6", "province") # polygon
问题
1.使用python bulk导入multipolygon时报错：Invalid LinearRing found. Found a single coordinate when expecting a coordinate array
sourcecode 638行
原因：
- 同一个geojson文件中存在存在类型为polygon和multipolygon的feature
- 设定mapping为polygon，插入multipolygon数据会报错：
invalid number of points in LinearRing (found [1] - must be >= [4])； - 设定mapping为multipolygon，插入polygon数据会报错：
Invalid LinearRing found. Found a single coordinate when expecting a coordinate array
解决方法：
判断feature的类型，创建的mapping不同，ES支持同一个type中同时存储polygon/multipolygon
