如何将Elasticsearch筛选出的完整文档存入Logstash事件？

直接使用fields属性无法动态获取结构多变的完整文档，因为它需要手动指定字段映射。这里提供两种可行的方案：

方案一：利用Script Fields + Ruby Filter处理 #

通过script_fields提取每个匹配文档的_source（完整文档内容），再用Ruby Filter将拼接的结果转换成数组存入results字段：

# 第一步：用script_fields获取完整文档的_source
  elasticsearch {
    query => '{"term": {"memberGuid": "%{[memberGuid]}"}}' # 修正查询语法为合法JSON
    index => "members-*"
    script_fields => {
      "full_document" => {
        "script" => "params._source"
      }
    }
    result_size => 1000
    add_field => { "[temp_docs]" => "%{[full_document]}" } # 临时存储拼接的文档内容
  }

  # 第二步：用Ruby Filter将临时字段转换为文档数组
  ruby {
    code => '
      temp_docs = event.get("[temp_docs]")
      if temp_docs
        # 拆分逗号分隔的文档字符串，解析为JSON对象并组成数组
        parsed_docs = temp_docs.split(",").map { |doc_str| JSON.parse(doc_str.strip) }
        event.set("results", parsed_docs)
      end
      # 删除临时字段
      event.remove("[temp_docs]")
    '
  }
}

注意事项 #

• 如果文档内容中包含未转义的逗号，拆分逻辑会出错，此时建议使用方案二。
• 确保Logstash已安装Ruby插件（默认已包含）。

方案二：直接调用Elasticsearch API（更稳定灵活） #

使用http filter直接调用ES的_search接口，获取完整搜索结果后提取文档数组：

# 调用Elasticsearch搜索API
  http {
    url => "http://your-es-host:9200/members-*/_search" # 替换为你的ES地址
    method => "POST"
    body => '{"query": {"term": {"memberGuid": "%{[memberGuid]}"}}, "size": 1000}' # 大小对应result_size
    headers => {
      "Content-Type" => "application/json"
    }
    target => "es_raw_response" # 将返回结果存入临时字段
    # 如果ES开启认证，添加以下配置
    # user => "your_username"
    # password => "your_password"
  }

  # 提取_source数组存入results字段
  ruby {
    code => '
      hits = event.get("[es_raw_response][hits][hits]")
      if hits
        # 从每个hit中提取_source（完整文档）
        results = hits.map { |hit| hit["_source"] }
        event.set("results", results)
      end
      # 删除临时的原始响应字段
      event.remove("[es_raw_response]")
    '
  }
}

方案优势 #

• 无需处理字段拼接问题，直接获取完整的搜索结果结构，适配任意索引结构。
• 可灵活添加认证、超时等HTTP配置，适配生产环境的ES集群。

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