作者来自 Elastic David Pilato在上一篇文章中我们介绍了如何将邮政地址转换为标准化地址并获取对应的地理坐标或者反过来根据地理坐标获取对应的邮政地址。现在假设我们已经有一个现成的数据集希望对其中的数据进行增强和补全。我们将考虑以下三种场景数据存储在 CSV 文件中。数据存储在关系型数据库 MySQL 中。数据已经存储在 Elasticsearch 中。接下来让我们看看如何对这些不同来源的数据进行增强。对 CSV 文件进行数据增强每当需要通过 Logstash 读取文件时我都非常喜欢使用 Filebeat 来完成这项工作。因此我修改了 Logstash 的输入部分不再使用http-input插件而是改用beats-input插件input { beats { port 5044 } }在filebeat.yml配置文件中我只进行了以下配置filebeat.prospectors: - type: log paths: - /path/to/data/*.csv close_eof: true output.logstash: hosts: [localhost:5044]此外我还启用了 X-Pack Monitoring监控功能以便了解 Pipeline数据处理管道的执行情况和性能表现xpack.monitoring.enabled: true xpack.monitoring.elasticsearch: hosts: [localhost:9200]我编写了一个简单的负载测试脚本如下所示。该脚本会对数据进行 10 轮处理cd filebeat* time for i in seq 1 10; do echo Launch $i rm data/registry ; ./filebeat --once done cd -以下是我使用的输入数据集$ wc -l data/person_dataset.csv 2499 data/person_dataset.csv因此数据集大约包含 2,500 行记录。数据内容如下3,Joe Smith,2000-11-15 23:00:00.000000,male,3,Paris,France,FR,47.26917867489252,-1.5316220472168889,44000 24,Nail Louisa,1980-05-02 22:00:00.000000,male,3,Nantes,France,FR,47.18584787904486,-1.6181576666034811,44000 36,Lison Nola,1985-09-23 22:00:00.000000,female,3,Nantes,France,FR,47.168657958748916,-1.5826229006751034,44000 45,Selena Sidonie,1964-10-18 23:00:00.000000,female,0,Paris,France,FR,48.82788569687699,2.2706737741614242,75000我们需要使用csv过滤器CSV Filter来解析这些数据csv { columns [id,name,dateOfBirth,gender,children,[address][city],[address][country],[address][countrycode],[location][lat],[location][lon],[address][zipcode]] convert { children integer [location][lat] float [location][lon] float } remove_field [host, version, timestamp,beat,source,tags,offset,prospector,message] }在这里由于输入数据包含地理坐标点Geo Location Points我们将采用上一篇文章中介绍的性能最慢的查询策略按地理距离排序Geo Distance Sorting。为了确保不会因为输出操作而过度拖慢 Pipeline数据处理管道的执行速度我将标准输出stdout的编解码器Codec替换为dotsoutput { stdout { codec dots } }完成 10 轮测试总共耗时3m3.842s约 3 分 4 秒。这意味着每处理 2,500 条文档并完成数据增强大约需要 18 秒相当于每秒处理约 140 条文档。这个性能还算不错。如果查看 Logstash 的监控数据可以看到每个事件的处理延迟Event Latency大约在 20–40 毫秒之间。Logstash 监控Logstash MonitoringLogstash 数据处理管道Logstash PipelineElasticsearch 过滤器插件在 Elasticsearch 中执行数据查找Lookup确实会降低整个处理流程的速度但我认为影响并不算太大平均每个事件耗时约 34 毫秒。对于 ETL提取、转换、加载操作来说这样的性能完全可以接受。这也是为什么在 Logstash 中执行耗时较长的操作通常比直接在 Elasticsearch 的 Ingest Pipeline摄取管道中执行更合适。因为 Ingest Pipeline 是在文档索引写入过程中调用的如果其中包含执行时间较长的操作就可能导致 Elasticsearch 的索引写入请求不断积压最终使索引写入队列逐渐被填满。连接其他数据源除了通过 Filebeat 读取 CSV 文件之外我们还可以从其他数据源读取数据。例如如果现有数据存储在 SQL 数据库中就可以使用jdbc-input插件直接读取数据库中的数据。相应的配置大致如下jdbc { jdbc_driver_library mysql-connector-java-6.0.6.jar jdbc_driver_class com.mysql.cj.jdbc.Driver jdbc_connection_string jdbc:mysql://127.0.0.1:3306/person?useSSLfalse 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-input插件连接到 Elasticsearch读取已经存储在某个索引中的现有数据并对其进行增强和补全。现在你已经掌握了实现类似地址转换和数据增强功能所需的全部工具。需要注意的是你可以使用任何可用的数据集并不局限于 BANO。我的下一步计划是将其他开放数据源导入 Elasticsearch尝试将地址数据的覆盖范围从法国扩展到更多国家。敬请期待原文Enriching your postal addresses with Elastic stack - part 3 | David Pilato
