Logstash:运用 Elasticsearch filter 来丰富地理数据

2022年12月12日   |   by mebius

我们知道丰富数据对于很多的应用来说非常重要。这涉及到访问不同的表格,并进行搜索匹配。找到最为相近的结果并进行丰富数据。针对 Elasticsearh 来说,我们可以通过 enrich processor来进行丰富。你可以阅读我之前的文章来了解更多:

事实上,我们甚至可以在 Logstash 的 pipeline 中采用 Elasticsearch filter 来丰富数据。你可以参考我之前的文章 “Logstash:运用 Elasticsearch 过滤器来丰富数据”。

在今天的文章中,我来采用一个简单的例子来展示如何使用 Elasticsearch filter 来丰富地理数据。这个在实际的使用中非常有用。比如你采集的数据含有传感器 id,但是在采集的数据里可能并不包含地理位置信息。我们可以通过 Elasticsearch filter 来丰富数据的地理位置信息,这样可以在地图上进行展示。

%title插图%num

安装

如果你还没有安装好自己的 Elasticsearch 及 Kibana,请参考文章:

在安装的时候,请参考 Elastic Stack 8.x 的安装指南进行安装。

在 Elasticsearch 中创建 enrich 索引

我们在 Elasticsearch 中创建一个可以在 Logstash 中被引用的索引。它的 mapping 是这样的:

PUT myindex
{
  "mappings": {
    "properties": {
      "address": {
        "properties": {
          "city": {
            "type": "text"
          },
          "number": {
            "type": "keyword"
          },
          "street_name": {
            "type": "text"
          },
          "zipcode": {
            "type": "keyword"
          }
        }
      },
      "location": {
        "type": "geo_point"
      }
    }
  }
}

我们接下来写入如下的一个文档:

POST myindex/_doc/1
{
  "address": {
    "zipcode": "17000",
    "number": "23",
    "city": "Beijing",
    "street_name": "wang jing road"
  },
  "location": {
    "lon": 116.478598,
    "lat": 39.995007
  }
}

从上面的数据中,我们可以看到除了含有 address 信息之前,它还含有一个地理位置的信息 location。在实际的使用中,那个 address 甚至可以是传感器的 id 信息。

如果我们知道 address 信息,那么我们可以通过如下的查询来获得该位置的地理信息:

GET myindex/_search?search_type=dfs_query_then_fetch&filter_path=**.hits
{
  "size": 1,
  "query": {
    "bool": {
      "should": [
        {
          "match": {
            "address.number": "23"
          }
        },
        {
          "match": {
            "address.street_name": "wang jing road"
          }
        },
        {
          "match": {
            "address.city": "Beijing"
          }
        }
      ]
    }
  }
}

上面的搜索返回如下的信息:

{
  "hits": {
    "hits": [
      {
        "_index": "myindex",
        "_id": "1",
        "_score": 1.4384104,
        "_source": {
          "address": {
            "zipcode": "17000",
            "number": "23",
            "city": "Beijing",
            "street_name": "wang jing road"
          },
          "location": {
            "lon": 116.478598,
            "lat": 39.995007
          }
        }
      }
    ]
  }tgcode
}

很显然,它含有 location 的地理位置信息。

在 Logstash 中使用 Elasticsearch filter 来丰富数据

接下来,我们创建如下的一个 Logstash 配置文件:

logstash.conf

input {
  http { }
}

filter {
  elasticsearch {
    query_template => "search-by-name.json"
    index => "myindex"
    fields => {
      "location" => "[location]"
      "address" => "[address]"
    }
    remove_field => ["headers", "host", "@version", "@timestamp"]
    ca_file => "/Users/liuxg/elastic/elasticsearch-8.5.2/config/certs/http_ca.crt"
    ssl => true
    api_key => "Rf-_4IQB-Ec0fhu3PjhI:JlZ0cA8lRQGRDxhWwdDJVg"
  }
}

output {
  stdout { codec => rubydebug }
}

如上所示,我们使用 http input 来做测试。在上面,我们使用 elasticsearch filter 来丰富数据。我们希望得到的是 location 及 address。在上面,我们定义 search-by-name.json 如下:

search-by-name.json

{
  "size": 1,
  "query":{
    "bool": {
      "should": [
        {
          "match": {
            "address.number": "%{[address][number]}"
          }
        },
        {
          "match": {
            "address.street_name": "%{[address][street_name]}"
          }
        },
        {
          "match": {
            "address.city": "%{[address][city]}"
          }
        }
      ]
    }
  }
}

在上面,我们设置 size 为 1,也即采用最为匹配的那个结果。上面其实就是一个搜索。我们把上述文件置于 Logstash 的安装目录下。另外,我们需要定义证书及 API key。你需要根据自己的配置来对上面的配置进行修改。如果你还不值得如何得到 API key,请阅读我之前的文章 “Logstash:如何连接到带有 HTTPS 访问的集群”。当然如果你的 Elasticsearch 不具有安全配置,那么你就不需要进行任何的配置了。

我们使用如下的命令来进行运行:

 ./bin/logstash -f logstash.conf 

%title插图%num

我们接下来在另外一个 terminal 中打入如下的命令:

curl -XPOST "localhost:8080" -H "Content-Type: application/json" -d '{
  "test_case": "Address with text",
  "name": "Joe Smith",
  "address": {
    "number": "23",
    "street_name": "wang jing road",
    "city": "Beijtgcodeing",
    "country": "China"
  }
}'

请注意在上面的命令中,它并没有 location 的任何信息。等执行完后,我们可以在 Logstash 的 terminal 中看到如下的输出:

%title插图%num

如上所示,我们可以看到被丰富的 location 信息。这个信息来源于 Elasticsearch 中的 myindex 索引。

现在你只需要改变输入部分来读取你的数据源,例如从数据库:

jdbc {
  jdbc_driver_library => "mysql-connector-java-6.0.6.jar"
  jdbc_driver_class =&gtgcodet; "com.mysql.cj.jdbc.Driver"
  jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/person?useSSL=false"
  jdbc_user => "root"
  jdbc_password => ""
  schedule => "* * * * *"
  parameters => { "country" => "France" }
  statement => "SELECT p.id, p.name, p.dateOfBirth, p.gender, p.children, a.city, a.country, a.countrycode, a.lat, a.lon, a.zipcode FROM Person p, Address a WHERE a.id = p.address_id AND a.country = :country AND p.id > :sql_last_value"
   use_column_value => true
   tracking_column => "id"
}

关于这个部分的详细操作请参考文章 “Elasticsearch:将关系数据库中的数据提取到 Elasticsearch 集群中”

文章来源于互联网:Logstash:运用 Elasticsearch filter 来丰富地理数据

相关推荐: Elasticsearch:使用反向地理编码在地图上显示自定义区域统计数据

在实际的许多应用中,我们可能并不一定按照行政区来进行划分区域,比如我们常说江浙一代,我们可以理解江苏和浙江这两个省合在一起,而不是把它们分开。我们有时也说长江三角区,它可能是跨几个省市的一个区域,而不是单纯的一个行政区。又或者,我们常说中原城市群这样的一个概念…

Tags: , , , ,