Elastic Stack 實現地理空間數據採集與可視化分析

隨著人類在不斷地探索空間，地理空間數據越來越多。收集信息的速度以及提供位置信息的來源正在迅速增長。政府和商業衛星繼續擴張。與GPS一起，它們提供了一系列不同的空間豐富的數據源，包括天氣和溫度模式，土地使用，土壤化學，減災和響應，電信等。

移動設備和底層網絡將人員，汽車，卡車和大量踏板車變成了位置信息的來源。計算機網絡將位置信息嵌入IP地址元數據中，這可以幫助IT管理員在分佈式基礎架構中為用戶提供支持，或者幫助執法部門以及我們的網絡運營商找到並阻止壞人。所有這些數據都是令人興奮的，它激發了每個人內部的創造力來利用它。提出新的問題，構思新的想法，並建立新的期望。這些新事物不容易解決。他們需要以不同格式存儲的數據或跨非空間維度（如主題標籤或網絡域）的相關性存儲的數據。儘管專家長期以來擁有執行復雜的地理空間分析的工具，但這些工具並不總是能夠完成非傳統來源的混合或處理當今數據集規模的任務。現在事情變得更加複雜。Elastic Stack 是一個高效的存儲，分析及搜索軟件棧。Elastic Stack 正在積極地推動這一進程。為我們提供更多的數據，更多的用途和更多的利益。

準備工作：

我將使用 opensky network API 來獲取飛機的飛行信息，並使用 Kibana 來對數據進行展示。

1、開通阿里雲 Elasticsearch 1核2G ，免費測試環境
2、開通阿里雲 Logstash 2核4G ，免費測試環境

我們通過阿里雲控制檯，進入阿里雲 kibana

Logstash 配置文件

在這個展示中，我們將使用 Logstash 定期地去 opensky network 網站上去抓取數據，並導入到 Elasticsearch 中。Logstash 的配置文件如下：

fligths_logstash.conf

input {
    http_poller {
        codec => "json"
        schedule => { every => "15s" }
        urls => {
            url => "https://opensky-network.org/api/states/all"
        }
    }
}
 
filter {
    split {
        field => "states"
        add_field => {
            "icao" => "%{[states][0]}"
            "callsign" => "%{[states][1]}"
            "origin_country" => "%{[states][2]}"
            "time_position" => "%{[states][3]}"
            "last_contact" => "%{[states][4]}"
            "location" => "%{[states][6]},%{[states][5]}"
            "baro_altitude" => "%{[states][7]}"
            "on_ground" => "%{[states][8]}"
            "velocity" => "%{[states][9]}"
            "true_track" => "%{[states][10]}"
            "vertical_rate" => "%{[states][11]}"
            "geo_altitude" => "%{[states][13]}"
            "squawk" => "%{[states][14]}"
            "spi" => "%{[states][15]}"
            "position_source" => "%{[states][16]}"
        }
    }
    mutate {
        strip => ["callsign"]
        rename => { "time" => "request_time" }
        remove_field => ["states", "@version"]
    }
    translate {
        field => "[position_source]"
        destination => "position_source"
        override => "true"
        dictionary => {
          "0" => "ADS-B"
          "1" => "ASTERIX"
          "2" => "MLAB"
        }
    }
 
    if [time_position] =~ /^%{*/ {
        drop { }
    }
    if [callsign] =~ /^%{*/ {
        mutate { remove_field => ["callsign"] }
    }
    if [location] =~ /^%{*/ {
        mutate { remove_field => ["location"] }
    }
    if [baro_altitude] =~ /^%{*/ {
        mutate { remove_field => ["baro_altitude"] }
    }
    if [velocity] =~ /^%{*/ {
        mutate { remove_field => ["velocity"] }
    }
    if [true_track] =~ /^%{*/ {
        mutate { remove_field => ["true_track"] }
    }
    if [vertical_rate] =~ /^%{*/ {
        mutate { remove_field => ["vertical_rate"] }
    }
    if [sensors] =~ /^%{*/ {
        mutate { remove_field => ["sensors"] }
    }
    if [geo_altitude] =~ /^%{*/ {
        mutate { remove_field => ["geo_altitude"] }
    }
    if [squawk] =~ /^%{*/ {
        mutate { remove_field => ["squawk"] }
    }
 
    mutate {
        convert => { 
            "baro_altitude" => "float" 
            "geo_altitude" => "float"
            "last_contact" => "integer"
            "on_ground" => "boolean"
            "request_time" => "integer"
            "spi" => "boolean"
            "squawk" => "integer"
            "time_position" => "integer"
            "true_track" => "float"
            "velocity" => "float"
            "vertical_rate" => "float"
        }
    }
}
 
output {
    stdout { 
        codec => rubydebug
    }
 
    elasticsearch {
        manage_template => "false"
        index => "flights"
        # pipeline => "flights_aircraft_enrichment"
         hosts => "localhost:9200"
    }
}

從上面的 input 部分我們可以看出來：

input {
    http_poller {
        codec => "json"
        schedule => { every => "15s" }
        urls => {
            url => "https://opensky-network.org/api/states/all"
        }
    }
}

這裡，我們使用 http_poller 每隔15秒去抓一次數據.如果大家想知道這個 API https://opensky-network.org/api/states/all 的內容是什麼，你可以直接在瀏覽器的地址欄中輸入這個地址，你及可以看出來是什麼樣的格式的信息。你可以使用工具 http://jsonviewer.stack.hu/ 對這個 JSON 格式的返回信息進行格式化，以便更好地分析它裡面的數據。這對在 filter 部分的各個過濾器的使用的理解是非常有幫助的。在 filter 的最後部分，我也對數據的類型進行了轉換以便更好地在 Kibana 中進行分析。

為了能夠使得我們的 flights 的數據類型和我們轉換的數據類型進行很好的匹配，我們必須在 Kibana 中對這個索引定義好它的 mapping：

PUT flights
{
  "mappings": {
    "properties": {
      "@timestamp": {
        "type": "date"
      },
      "baro_altitude": {
        "type": "float"
      },
      "callsign": {
        "type": "keyword"
      },
      "geo_altitude": {
        "type": "float"
      },
      "icao": {
        "type": "keyword"
      },
      "last_contact": {
        "type": "long"
      },
      "location": {
        "type": "geo_point"
      },
      "on_ground": {
        "type": "boolean"
      },
      "origin_country": {
        "type": "keyword"
      },
      "position_source": {
        "type": "keyword"
      },
      "request_time": {
        "type": "long"
      },
      "spi": {
        "type": "boolean"
      },
      "squawk": {
        "type": "long"
      },
      "time_position": {
        "type": "long"
      },
      "true_track": {
        "type": "float"
      },
      "velocity": {
        "type": "float"
      },
      "vertical_rate": {
        "type": "float"
      }
    }
  }
}

在 Kibana 的 Dev Tools 中運行上面的指令，這樣我們就創建好了 flights 索引的 mapping。

接下來，我就可以啟動通過阿里雲控制檯，進入Logstash 了。

這時，我們可以在 console 中看到如下的輸出：