使用弹性搜索地理功能从 XML 原始数据中查找最常见的位置

Question

我想使用 Elastic Search 及其地理功能来生成最常见位置的排序列表.

其中许多位置将是相同的物理位置（例如用户的家），但显然经度和纬度可能不完全相同。

考虑到一周中每一天的数据，在整个数据周期（将是一个月）内应该是一周中的同一天。例如，在恰好是星期二的日期搜索公共位置，我们应该查询上星期二、前星期二、前星期二和前星期二的数据（！）[也许这可以通过使用 ES 来实现索引？]。

对于每个搜索日，我还想要一个直方图，准确度为 15 分钟，再次显示他们在该位置 100m 范围内的所有时间，包括一周中同一天的最后 4 周数据。

我们只能在一天内发出 API 原始数据请求（因此需要多次请求才能获取最近 4 周的数据）。不受我们控制的第三方 API 将以以下格式返回 XML - 全部在一行中且未格式化（我手动格式化了下面的示例）。坐标是（经度，纬度）格式。最后一个数字（下例中的 0）代表海拔高度，如果可能，应该存储。

<?xml version="1.0" encoding="UTF-8"?>
<kml>
   <Document>
      [stuff we don't care about]
      <Day>
         [stuff we don't care about]
         <Locations>
        [stuff we don't care about]
            <time>2016-04-30T19:35:01.558+10:00</time>
            <coord>142.9987247 -37.328203799999996 0</coord>
            <time>2016-05-02T12:29:21.233+10:00</time>
            <coord>142.96122699999998 -37.921569999999996 0</coord>
            ....
         </Locations>
      </Day>
   </Document>
</kml>

非常感谢。

Answer 1

与您的other question 类似，可以很容易地解析给定的 XML 并将结果位置索引到 elasticsearch 中。为了提取数据，需要进行一些 XML 解析，然后执行一些数据按摩，但这是可能的。

我想出了下面非常简单的 Logstash 配置：

input {
  http_poller {
    urls => {
      get_locations => {
        method => get
        url => "http://your-api.com/locations.xml"
        headers => {
          Accept => "application/xml"
        }
      }
    }
    request_timeout => 60
    interval => 60
    codec => "plain"
  }
}
filter {
  # 1. parse XML
  xml {
    source => "message"
    force_array => false
    target => "parsed"
  }

  # 2. parse time/coord arrays and rebuild pairs
  ruby {
    code => "
      event['locations'] = []
      event['parsed']['Document']['Day']['Locations']['time'].each { |time|
        event['locations'].push({'time' => time, 'location' => nil})
      }
      event['parsed']['Document']['Day']['Locations']['coord'].each_with_index { |coord, i|
        event['locations'][i]['location'] = {
          'lon' => coord.split(' ')[0],
          'lat' => coord.split(' ')[1]
        }
      }
    "
  }

  # 3. produce one event per time/coord pair
  split {
    field => "locations"
  }

  # 4. Some renaming and clean-ups    
  mutate {
    rename => {
      "[locations][time]" => "timestampMs"
      "[locations][location]" => "location"
    }
    remove_field => [
      "parsed", "message", "@timestamp", "@version", "locations"
    ]
  }
}
output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "locations"
    document_type => "location"
  }
}

首先，我使用http_poller 输入来提取 XML 数据

然后，我使用 xml 过滤器将 XML 解析为 JSON。您提供的 XML 将生成以下 JSON：

 {
    "Document" => {
        "Day" => {
            "Locations" => {
                 "time" => [
                    [0] "2016-04-30T19:35:01.558+10:00",
                    [1] "2016-05-02T12:29:21.233+10:00"
                ],
                "coord" => [
                    [0] "142.9987247 -37.328203799999996 0",
                    [1] "142.96122699999998 -37.921569999999996 0"
                ]
            }
        }
    }

如您所见，由于笨拙的 XML 组织，time 和 coord 值都在各自的数组中粘合在一起。

然后，我利用ruby 过滤器将其全部拆分，并将每个time 与其适当的coord 值重新结合。我基本上遍历每个数组并重新构造正确的time/coord 对并将它们存储到一个新的locations 数组中。请注意，当前版本的 Elasticsearch 中的高度部分为 not yet supported。

然后我 split 那个新的 locations 数组以便为每个 time/coord 对产生一个事件。

最后我正在做一些清理工作，将被索引到 Elasticsearch 中的事件如下所示：

{
 "timestampMs" => "2016-04-30T19:35:01.558+10:00",
    "location" => {
         "lon" => "142.9987247",
         "lat" => "-37.328203799999996"
    }
}
{
 "timestampMs" => "2016-05-02T12:29:21.233+10:00",
    "location" => {
         "lon" => "142.96122699999998",
         "lat" => "-37.921569999999996"
    }
}

然后您可以运行 bin/logstash -f locations.conf 以运行您的管道。

有了它，您就可以重用与 other question 中相同的聚合，这将起作用。