Observability：使用 Elastic Stack 分析地理空間數據

在之前的文章 “Observability：使用 Elastic Stack 分析地理空間數據（一）”，我詳述瞭如何從 OpenSky Network API 接口把數據導入到 Elasticsearch，並對這些數據進行可視化分析。也許針對很對的情況這個已經很滿足了，因為它確實可以幫我們從很多實時數據中提取很多有用的東西。

在今天的文章中，我們將參考之前的文章 “如何使用 Elasticsearch ingest 節點來豐富日誌和指標” 。我們可以利用 Elasticsearch ingest 節點來更加豐富我們的數據，並對這些數據做更進一步的的分析。

為了達到這個目的，我們必須首先了解在之前索引中的 icao 字段。這個字段的意思是：

ICAO 機場代碼或位置指示器是由四個字母組成的代碼，用於指定世界各地的機場。這些代碼由國際民用航空組織定義併發布在國際民航組織7910號文件：位置指示器中，供空中交通管制和航空公司運營（例如飛行計劃）使用。

我們之前的每個文檔是這樣的：


{
  "velocity" : 0.0,
  "icao" : "ad0851",
  "true_track" : 264.38,
  "time_position" : 1591190152,
  "callsign" : "AAL2535",
  "origin_country" : "United States",
  "position_source" : "ADS-B",
  "spi" : false,
  "request_time" : 1591190160,
  "last_contact" : 1591190152,
  "@timestamp" : "2020-06-03T13:16:03.723Z",
  "on_ground" : true,
  "location" : "32.7334,-117.2035"
}

另外，我們可以在地址 https://opensky-network.org/datasets/metadata/ 找到一個如下文件：

在這裡，我們可以找到一個叫做 aircraftDatabase.csv 的文件。它裡面的內容如下：

在上面的表格中，我們發現有一個叫做 icao24 的字段。這個字段和我們之前的文檔可以進行關聯，從而我們可以得到更多關於某個航班的更多信息。

創建 enrich index

由於下載的文檔時一個是一個 csv 的文件。我們可以使用 data visualizer 來導入。

點擊上面的 Override settings 鏈接：

點擊 Apply 按鈕：

點擊上面的 Import 按鈕：

我們把這個索引的名字稱作為 aircraft。點擊 Advaned:

再次確認 mapping，如果沒有問題的話，點擊 Import 按鈕:

由於這個文件比較大，所以需要一點時間來進行導入：

等完成後，我們可以在 Elasticsearch 中找到一個叫做 aircraft 的索引：

上面顯示有一個新的 aircraft 的索引生成了。

創建 Enrich policy

接下來，我們來創建 enrich policy。它告訴我們如何豐富數據。在 Kibana 中打入如下的命令：


PUT /_enrich/policy/flights_policy
{
  "match": {
    "enrich_fields": [
      "acars",
      "adsb",
      "built",
      "category_description",
      "engines",
      "first_flight_date",
      "icao_aircraft_type",
      "line_number",
      "manufacturer_icao",
      "manufacturer_name",
      "model",
      "modes",
      "notes",
      "operator",
      "operator_callsign",
      "operator_iata",
      "operator_icao",
      "owner",
      "reg_until",
      "registered",
      "registration",
      "seat_configuration",
      "serial_number",
      "status",
      "test_reg",
      "type_code"
    ],
    "indices": [
      "aircraft"
    ],
    "match_field": "icao"
  }
}

我們使用 execute enrich policy API 為該策略創建enrich索引：

POST /_enrich/policy/flights_policy/_execute

接著，我們創建一個叫做 flights_aircraft_enrichment 的 pipeline：


PUT /_ingest/pipeline/flights_aircraft_enrichment
{
  "description": "joins incoming ADSB state info with richer aircraft metadata",
  "processors": [
    {
      "enrich": {
        "field": "icao",
        "policy_name": "flights_policy",
        "target_field": "aircraft"
      }
    }
  ]
}

到此為止，我們已經成功地創建了豐富策略。接下來，我們將展示如何使用這個 pipeline 來豐富我們的數據。

豐富數據

為了能夠使用我們上面定義好的 pipeline，我們重參考之前的文章 “Observability：使用 Elastic Stack 分析地理空間數據（一）”裡的 fligths_logstash.conf 文件，並修改如下的 output 部分：


output {
    stdout { 
        codec => rubydebug
    }
 
    elasticsearch {
        manage_template => "false"
        index => "flights"
        # pipeline => "flights_aircraft_enrichment"
         hosts => "localhost:9200"
    }
}

我們把上面的這一行的註釋拿掉：

# pipeline => "flights_aircraft_enrichment"

這樣變成了：

output {
    stdout { 
        codec => rubydebug
    }
 
    elasticsearch {
        manage_template => "false"
        index => "flights"
        pipeline => "flights_aircraft_enrichment"
         hosts => "localhost:9200"
    }
}

在啟動 Logstash 之前，我們可以先刪除之前的 flights 索引：

DELETE flights

再接著執行如下的命令：

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"
      }
    }
  }
}

重新運行 Logstash：

sudo ./bin/logstash -f fligths_logstash.conf

我們在 Kibana 中檢查 flights 的 mapping:

GET flights/_mapping

我們可以看到一些新增加的各個新字段：

我們可以通過 search：

  "_source" : {
          "aircraft" : {
            "owner" : "Wells Fargo Trust Co Na Trustee",
            "reg_until" : "2021-04-30",
            "modes" : false,
            "built" : "1984-01-01",
            "acars" : false,
            "manufacturer_icao" : "BOEING",
            "serial_number" : "23018",
            "manufacturer_name" : "Boeing",
            "icao_aircraft_type" : "L2J",
            "operator_callsign" : "GIANT",
            "operator_icao" : "GTI",
            "engines" : "GE CF6-80 SERIES",
            "icao" : "a8a763",
            "registration" : "N657GT",
            "model" : "767-281",
            "type_code" : "B762",
            "adsb" : false
          },
          "true_track" : 272.81,
          "velocity" : 5.14,
          "spi" : false,
          "origin_country" : "United States",
          "@timestamp" : "2020-06-04T10:41:00.558Z",
          "request_time" : 1591267250,
          "time_position" : 1591267168,
          "last_contact" : 1591267168,
          "callsign" : "GTI165",
          "icao" : "a8a763",
          "location" : "39.0446,-84.6505",
          "on_ground" : true,
          "position_source" : "ADS-B"
        }
      }

我們可看到一個叫做 aircraft 的字段，它含有這個飛機所有被豐富的信息。