сопоставление geo_point python и сбой StreamSets с Elasticsearch

Question

У меня есть это отображение вasticsearch

"mappings": {
          "properties": {
                "fromCoordinates": {"type": "geo_point"},
                "toCoordinates": {"type": "geo_point"},
                "seenCoordinates": {"type": "geo_point"},
            }
        }

С консолью кибаны не возникает проблем со всеми возможными комбинациями полей geo_ip , поддерживаемыми эластичным поиском , то есть:

(широта, долгота)

PUT /anindex/_doc/1
{
   "fromCoordinates": {
     "lat": 36.857200622558594    
     "lon": 117.21600341796875,

  },
  "toCoordinates": {
    "lat": 22.639299392700195    
    "lon": 113.81099700927734,

  },
  "seenCoordinates": {
     "lat": 36.91663    
     "lon": 117.216,
   }
}

(широта, широта)

PUT /anindex/_doc/2
{
 "fromCoordinates": [36.857200622558594, 117.21600341796875], 
 "toCoordinates": [22.639299392700195, 113.81099700927734], 
 "seenCoordinates": [36.91663, 117.216] 
}

Но я попытался вставить в эластичный поиск данные через python, и у меня всегда есть эта ошибка:

RequestError(400, 'illegal_argument_exception', 'mapper [fromCoordinates] of different type, current_type [geo_point], merged_type [ObjectMapper]')

В python я строю json из словаря, и это результат, когда я печатал:

fromCoordinates = {}
fromCoordinates['lat'] = fromLat  
fromCoordinates['lon'] = fromLon 

dataDictionary.update({'fromCoordinates': fromCoordinates , 'toCoordinates': toCoordinates, 'seenCoordinates': seenCoordinates})

print(json.dumps(dataDictionary).encode('utf-8'))

{"fromCoordinates": {"lat": 43.9962005615, "lon": 125.684997559}, 
"toCoordinates": {"lat": 40.080101013183594, "lon": 116.58499908447266}, 
"seenCoordinates": {"lat": 33.62672, "lon": 109.37243}}

и загрузить с этим

data = json.dumps(dataDictionary).encode('utf-8')
es.create(index='anindex', doc_type='document', id=0, body=data)

Версия массива имеет те же проблемы:

fromCoordinates = [fromLon, fromLat]

Это json, созданный и напечатанный в python:

{"fromCoordinates": [113.81099700927734, 22.639299392700195], 
  "toCoordinates": [106.8010025024414, 26.53849983215332], 
   "seenCoordinates": [107.46743, 26.34169]}

В этом случае у меня есть этот ответ

RequestError: RequestError(400, 'mapper_parsing_exception', 'geo_point expected')

Та же ошибка возникает, если я пытаюсь использовать StreamSets дляasticsearch, имея оба типа json, показанные ранее:

mapper [fromCoordinates] of different type, current_type [geo_point], merged_type [ObjectMapper]

Есть идеи?

ОБНОВЛЕНИЕ:

GET /anindex/_mapping

{ "anindex" : 
   { "mappings" : 
     { "properties" : 
       { "fromCoordinates" : 
          { "type" : "geo_point" }, 
        "toCoordinates" : 
           { "type" : "geo_point" }, 
        "seenCoordinates" : { "type" : "geo_point" } 
       }
      }
    }
 }

РЕШЕНИЕ:

После примера, приведенного @jzz я понял, что параметр doc_type в es.create(index='anindex', doc_type='document', id=0, body=data) вызывает ошибку, я удалил ее, и она сработала ..... Но мне все еще интересно, почему в StreamSets такая же ошибка ... но я продолжу с python.

Answer 1

Если вы использовали более старую версиюasticsearch (например, 6.1) и обновили ее до более новой версии (например, 7.X) - вам нужно удалить doc_type в шаблоне индексации, поскольку более новая версия больше не принимает этот объект.

старый шаблон индексации

res=es_local.index(index='local-index',doc_type='resource', body=open_doc,id=_id,request_timeout=60)

новый шаблон индексации

res=es_local.index(index='local-index', body=open_doc,id=_id,request_timeout=60)

Примечание: - в новом шаблоне индексации нет doc_type (предполагается индексирование с использованием python).

Answer 2

Я подозреваю, что сначала вы установили отображение object на fromCoordinates, а затем попытались обновить отображение. Попробуйте сбросить и воссоздать индекс, и тогда все эти варианты будут работать нормально:

---

Python

from elasticsearch import Elasticsearch
import time

es_instance = Elasticsearch(['http://localhost:9200'])

es_instance.indices.create(
    'anindex',
    body={"mappings": {
        "properties": {
            "fromCoordinates": {"type": "geo_point"},
            "toCoordinates": {"type": "geo_point"},
            "seenCoordinates": {"type": "geo_point"}
        }
    }})

es_instance.create(
    index="anindex",
    id=0,
    body={
        "fromCoordinates": {"lat": 43.9962005615, "lon": 125.684997559},
        "toCoordinates": {"lat": 40.080101013183594, "lon": 116.58499908447266},
        "seenCoordinates": {"lat": 33.62672, "lon": 109.37243}})

es_instance.create(
    index="anindex",
    id=1,
    body={
        "fromCoordinates": [
            117.21600341796875,
            36.857200622558594
        ],
        "toCoordinates": [
            113.81099700927734,
            22.639299392700195
        ],
        "seenCoordinates": [
            117.216,
            36.91663
        ]
    })

# syncing is not instant so wait
time.sleep(1)

print(es_instance.count(index="anindex"))

---

Кибана:

DELETE anindex

PUT anindex
{
  "mappings": {
    "properties": {
      "fromCoordinates": {
        "type": "geo_point"
      },
      "toCoordinates": {
        "type": "geo_point"
      },
      "seenCoordinates": {
        "type": "geo_point"
      }
    }
  }
}

PUT /anindex/_doc/1
{
  "fromCoordinates": {
    "lat": 36.857200622558594,
    "lon": 117.21600341796875
  },
  "toCoordinates": {
    "lat": 22.639299392700195,
    "lon": 113.81099700927734
  },
  "seenCoordinates": {
    "lat": 36.91663,
    "lon": 117.216
  }
}

PUT /anindex/_doc/2
{
  "fromCoordinates": [
    117.21600341796875,
    36.857200622558594
  ],
  "toCoordinates": [
    113.81099700927734,
    22.639299392700195
  ],
  "seenCoordinates": [
    117.216,
    36.91663
  ]
}

PUT anindex/_doc/3
{
  "fromCoordinates": "22.639299392700195,113.81099700927734",
  "toCoordinates": "26.53849983215332,106.8010025024414",
  "seenCoordinates": "26.34169,107.46743"
}