python - 为什么在数据流传输时映射不起作用？

Question

我需要将经纬度映射到地理点。我每隔10秒从网站流一次数据。我只从json第一次获得该位置。我做了映射。当我运行代码时，数据正在流式传输，但是看不到Kibana中的映射。

这是我的代码。
es = Elasticsearch('http://ip:port'，timeout = 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卫星定位/ _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

与其在Python脚本中添加该逻辑，不如使用ingest pipeline为您完成工作。

首先，在Kibana中运行以下调用以创建geo-pipeline提取管道，该管道将从您从Satellite 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字段，您可以在 map 上显示该字段。