xml - 使用 Elasticsearch 地理功能从 XML 原始数据中查找最常见的位置

Question

我想使用 Elastic Search 和它的地理功能来生成最常见位置的排序列表，如果它们在一周中的某一天彼此相距 100m，则这些位置被认为是相同的。

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

为一周中的每一天考虑的数据应该是整个数据周期(将是一个月)的一周中的同一天。例如，在恰好是星期二的日期搜索公共(public)位置，我们应该查询上星期二、前星期二、前星期二和前星期二的数据(!)[也许这可以通过使用 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值(value)。我基本上遍历每个数组并重新构造正确的 time/coord对并将它们存储到一个新的 locations大批。注意海拔部分是 not yet supported在当前版本的 Elasticsearch 中。

然后我 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 中相同的聚合。这将起作用。