如何用Logstash拆分Elasticsearch中OSM数据的other_tags字段为多字段

需要安装的工具 #

• Logstash：核心处理工具，负责数据的读取、转换与写入
• （已部署的）Elasticsearch：需确保与Logstash大版本一致（比如同属7.x或8.x系列），避免兼容性问题
• 可选：Kibana，用于快速验证转换后的数据结构与内容

操作方式说明 #

无需直接连接ES执行命令或调用API，而是通过Logstash搭建数据处理管道：从ES的points索引读取原始数据，拆分other_tags字段后写入新的ES索引（建议先写入新索引验证，再替换原索引）。

完整实现步骤 #

1. 确认版本兼容 #

先查询Elasticsearch版本（可通过GET /请求查看），再安装对应大版本的Logstash，比如ES为7.17.x，就安装Logstash 7.17.x系列。

2. 编写Logstash配置文件 #

创建配置文件（例如命名为osm-split.conf），包含input（读数据）、filter（转数据）、output（写数据）三个核心模块：

输入模块：从ES读取原数据

input {
  elasticsearch {
    hosts => ["http://你的ES服务器地址:9200"] # 替换为实际ES地址
    index => "points" # 原索引名称
    query => '{ "query": { "match_all": {} } }' # 读取索引内所有数据
    scroll => "5m" # 批量读取超时时间，数据量大时可适当调长
    docinfo => true # 保留原文档ID等元数据
  }
}

过滤模块：拆分other_tags字段

根据你的other_tags实际格式选择对应配置：

情况A：other_tags为JSON格式字符串

例：other_tags值为{"city":"成都","housenumber":"101","country":"中国"}

filter {
  # 将JSON字符串解析为嵌套字段
  json {
    source => "other_tags"
    target => "address_temp"
  }
  # 提取需要的地址字段到根级别
  mutate {
    add_field => {
      "city" => "%{[address_temp][city]}"
      "housenumber" => "%{[address_temp][housenumber]}"
      "country" => "%{[address_temp][country]}"
      # 按需添加其他字段，如postcode、street等
    }
    # 可选：删除临时字段与原other_tags，减少数据体积
    remove_field => ["address_temp", "other_tags"]
  }
}

情况B：other_tags为OGR导出的键值对格式

例：other_tags值为'city=>成都, housenumber=>101, country=>中国'

filter {
  # 解析键值对格式字符串
  kv {
    source => "other_tags"
    field_split => ", "
    value_split => "=>"
    target => "address_temp"
  }
  # 提取地址字段到根级别
  mutate {
    add_field => {
      "city" => "%{[address_temp][city]}"
      "housenumber" => "%{[address_temp][housenumber]}"
      "country" => "%{[address_temp][country]}"
    }
    remove_field => ["address_temp", "other_tags"]
  }
}

输出模块：写入新索引

output {
  elasticsearch {
    hosts => ["http://你的ES服务器地址:9200"]
    index => "points_processed" # 新索引名称，避免覆盖原数据
    document_id => "%{[@metadata][_id]}" # 保留原文档ID，防止数据重复
  }
  # 可选：输出到控制台，实时查看转换结果
  stdout {
    codec => rubydebug
  }
}

3. 运行Logstash执行转换 #

打开终端，进入Logstash安装目录，执行命令：

# Linux/macOS系统
./bin/logstash -f /path/to/osm-split.conf

# Windows系统
bin\logstash.bat -f C:\path\to\osm-split.conf

等待Logstash处理完所有数据（终端会显示处理进度）。

4. 验证转换结果 #

通过Kibana Dev Tools或curl查询新索引：

GET /points_processed/_search
{
  "query": { "match_all": {} },
  "_source": ["city", "housenumber", "country"]
}

确认返回结果中已生成独立的地址字段。

5. 替换原索引（可选） #

若验证无误，可将新索引数据导回原索引：

# 用_reindex将数据同步回原索引
POST /_reindex
{
  "source": { "index": "points_processed" },
  "dest": { "index": "points" }
}

# 删除临时索引
DELETE /points_processed

内容的提问来源于stack exchange，提问作者heikorm