Logstash基于公共ID关联多索引生成新索引的问题求助

问题根源 #

现有配置未对index2的匹配结果做校验，无匹配记录时仍会将index1原始数据输出至新索引，同时缺少index3关联逻辑，需先确认查询模板的关联条件是否正确。

步骤1：配置正确的查询模板 #

index2关联查询模板（query_template_forindex.json） #

{
  "query": {
    "term": {
      "refId": "{{index1refId}}"
    }
  },
  "size": 1
}

index3关联查询模板（query_template_forindex3.json） #

{
  "query": {
    "term": {
      "ref3Id": "{{index1refId}}"
    }
  },
  "size": 1
}

如果关联字段为text类型，需将查询字段改为对应.keyword子字段

步骤2：修改Logstash配置 #

input{
   elasticsearch{
   hosts => "hostname"
   index => "index1"
   query => '{"query": {"match_all":{}}}' 
   docinfo => true
   user => "uname"
   password => "pwd"
   ssl => true
 }
}
filter{
   # 关联index2
   elasticsearch{
     hosts =>"hostname"
     enable_sort => false
     index => "index2"
     user => "uname"
     password => "pwd"
     query_template => "query_template_forindex.json" 
     fields => {
      "name" => "name"
      "sal" => "sal"
      "date" => "date" 
      }
      tag_on_failure => ["_no_index2_match"]
   }
   # 关联index3
   elasticsearch{
     hosts =>"hostname"
     enable_sort => false
     index => "index3"
     user => "uname"
     password => "pwd"
     query_template => "query_template_forindex3.json" 
     fields => {
      "dob" => "dob"
      }
      tag_on_failure => ["_no_index3_match"]
   }
   # 统一公共ID字段，清理冗余字段
   mutate {
     add_field => { "refID" => "%{index1refId}" }
     remove_field => ["index1refId", "@timestamp", "@version"]
   }
   # 丢弃未同时匹配三个索引的记录
   if "_no_index2_match" in [tags] or "_no_index3_match" in [tags] {
     drop {}
   }
}
output{
  elasticsearch{
     hosts => "hostname"
     index => "newindex"
     user => "uname"
     password => "pwd"
     doc_as_upsert => true
     document_id => "%{refID}"
     ssl => true 
 }
  stdout { codec => rubydebug }
}

配置说明 #

• tag_on_failure参数会在对应索引无匹配记录时为事件打标记，通过条件判断直接丢弃未全匹配的记录，彻底避免多余未匹配数据进入最终索引
• 每个查询模板指定size: 1，避免单ID匹配到多条记录导致字段覆盖异常
• 输出结构完全符合预期要求，公共ID统一为refID，包含三个索引的所有关联字段

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