Почему отображение не работает при потоковой передаче данных?

Question

Мне нужно сопоставить широту и долготу географической точки. Я транслирую данные с веб-сайта каждые 10 секунд. Я получаю только первое вхождение позиции из json. Я сделал карту. и когда я запускаю свой код, данные передаются в потоковом режиме, но я не вижу сопоставления в Kibana.

Вот мой код. es = Elasticsearch ('http://ip: порт ', тайм-аут = 600)

settings = { "settings": {
                     "number_of_shards":1

                     },
          "mappings" : {
                    "properties":{
                         "geo": {
                            "properties": {
                                "location":{
                                   "type": "geo_point"

                                    }
                                  }
                        }
        } } }

es.indices.create(index = "myindex", ignore = 400, body=settings)
def data_collect():
 data = requests.get(url = URL).json() 
 del data["positions"][1]
 positions = data['positions']

 if "satlatitude" in positions and "satlongitude" in positions:
  data['positions']['geo'] = { 'location':      
                           str(positions['satlatitude'])+","+str(positions['satlongitude'])}

  es.index(index='myindex',doc_type='mydoc',body=data)

schedule.every(10).seconds.do( data_collect)
while True:
 schedule.run_pending()
 time.sleep(1)

Результаты GET Satellitepositions / _mapping, как показано ниже:

 {
 "satellitepositions": {
 "mappings": {
  "satelitepos": {
    "properties": {
      "info": {
        "properties": {
          "satid": {
            "type": "long"
          },
          "satname": {
            "type": "text",
            "fields": {
              "keyword": {
                "type": "keyword",
                "ignore_above": 256
              }
            }
          },
          "transactionscount": {
            "type": "long"
          }
        }
      },
      "positions": {
        "properties": {
          "azimuth": {
            "type": "float"
          },
          "dec": {
            "type": "float"
          },
          "eclipsed": {
            "type": "boolean"
          },
          "elevation": {
            "type": "float"
          },
          "ra": {
            "type": "float"
          },
          "sataltitude": {
            "type": "float"
          },
          "satlatitude": {
            "type": "float"
          },
          "satlongitude": {
            "type": "float"
          },
          "timestamp": {
            "type": "long"
          }
        }
      }
    }
    }
    }
    }
    }

Результаты позиций спутников GET, как показано ниже:

 {
 "satellitepositions": {
 "aliases": {},
  "mappings": {
  "satelitepos": {
    "properties": {
      "info": {
        "properties": {
          "satid": {
            "type": "long"
          },
          "satname": {
            "type": "text",
            "fields": {
              "keyword": {
                "type": "keyword",
                "ignore_above": 256
              }
            }
          },
          "transactionscount": {
            "type": "long"
          }
        }
      },
      "positions": {
        "properties": {
          "azimuth": {
            "type": "float"
          },
          "dec": {
            "type": "float"
          },
          "eclipsed": {
            "type": "boolean"
          },
          "elevation": {
            "type": "float"
          },
          "ra": {
            "type": "float"
          },
          "sataltitude": {
            "type": "float"
          },
          "satlatitude": {
            "type": "float"
          },
          "satlongitude": {
            "type": "float"
          },
          "timestamp": {
            "type": "long"
          }
        }
      }
    }
  }
},
"settings": {
  "index": {
    "creation_date": "1590738791943",
    "number_of_shards": "5",
    "number_of_replicas": "1",
    "uuid": "HLstIPiXQcyJC5_laowxNQ",
    "version": {
      "created": "6040399"
    },
    "provided_name": "satellitepositions"
  }
  }
  }
  }

Answer 1

Вместо добавления этого logi c в сценарий Python, давайте воспользуемся конвейером приема , который сделает эту работу за вас.

Сначала выполните следующий вызов в Kibana для создания конвейера захвата geo-pipeline, который будет создавать географическую точку из полей широты и долготы, которые вы получаете из спутникового API.

PUT _ingest/pipeline/geo-pipeline
{
  "processors": [
    {
      "script": {
        "source": """
          ctx.positions = ctx.positions[0];
          ctx.geo = [
            'lat': ctx.positions.satlatitude,
            'lon': ctx.positions.satlongitude
          ]
          """
      }
    }
  ]
}

Затем вы можете упростить свой Python код, как показано ниже . Вы заметите, что я изменил настройки индекса, чтобы убедиться, что все документы, которые проиндексированы в вашем индексе, сначала проходят через geo-pipeline, а поле geo создается правильно.

es = Elasticsearch('http://ip:port',timeout=600)

# create index
settings = {
  "settings": {
    "index.number_of_shards": 1,
    "index.default_pipeline": "geo-pipeline"
  },
  "mappings": {
    "satelitepos": {
      "properties": {
        "geo": {
          "properties": {
            "location": {
              "type": "geo_point"
            }
          }
        }
      }
    }
  }
}
es.indices.create(index = "satellitepositions", ignore = 400, body=settings)

# routine for collecting data
def data_collect():
  data = requests.get(url = URL).json() 
  del data["positions"][1]
  es.index(index='satellitepositions', doc_type='satelitepos', body=data)

schedule.every(10).seconds.do(data_collect)

# kick it all off
while True:
 schedule.run_pending()
 time.sleep(1)

Это должно работать так, как вы ожидаете, и ваши документы получат поле geo типа geo_point, которое вы можете отобразить на карте.