如何通过Logstash将Kafka数据传入REST API的HTTP请求体？

当然可行，你的需求可以通过Logstash的正确配置实现，先指出你现有配置里的几个关键问题：

• Input块语法错误：elasticsearch和kafka插件不能嵌套，input下每个数据源插件是平级配置的
• 错误使用http filter：http filter的作用是调用API拉取数据并添加到当前事件，而不是把当前事件数据推送到API。要推送事件数据到外部API，应该用http output插件

场景1：从Kafka获取数据，发送到REST API（同时可选存入Elasticsearch） #

如果你的核心需求是从Kafka拿到浮点数值，塞进REST API的请求体，同时把结果存入ES，配置应该这样写：

input {
  # 直接用kafka input获取数据，不需要嵌套elasticsearch
  kafka {
    bootstrap_servers => 'kafka:9094'
    topics => ["你的Kafka主题名"]
    client_id => "自定义客户端ID"
    group_id => "消费组ID"
    codec => json {}
  }
}

filter {
  # 如果需要对Kafka来的浮点数据做处理，比如提取字段、格式转换，在这里添加filter
  # 假设Kafka传来的字段是"float_value"，要放到API请求体的"instances"字段里
  mutate {
    add_field => {
      # 这里可以直接引用Kafka事件里的字段，比如%{float_value}
      "api_body_instances" => "%{float_value}"
    }
  }
}

output {
  # 第一个output：发送数据到REST API
  http {
    verb => "POST"
    url => "http://你的API地址"
    body_format => "json"
    # 这里直接构造请求体，引用filter里处理好的字段或者原始Kafka字段
    body => {
      "instances" => "%{api_body_instances}"
      # 如果需要其他字段，直接添加即可
      # "other_field" => "%{original_field}"
    }
    # 可选：如果API有认证，添加headers配置
    # headers => {
    #   "Authorization" => "Bearer 你的令牌"
    # }
  }

  # 第二个output：把数据存入Elasticsearch
  elasticsearch {
    hosts => ["elasticsearch:9200"]
    index => "demo_results"
  }
}

场景2：从Elasticsearch查询数据，发送到REST API #

如果你的需求是先从ES查询数据，再把这些数据发送到API，那需要用elasticsearch input插件来查询ES，再用http output发送：

input {
  elasticsearch {
    hosts => ["elasticsearch:9200"]
    # 配置查询条件，比如查询demo_results索引里的所有数据
    index => "demo_results"
    query => '{ "query": { "match_all": {} } }'
    # 可选：设置查询间隔，比如每1分钟查一次
    schedule => "* * * * *"
    codec => json {}
  }
}

filter {
  # 对ES查询出来的数据做处理，比如提取需要的字段
  mutate {
    # 假设ES里的目标字段是"float_value"，准备放到API请求体
    add_field => {
      "api_instances" => "%{float_value}"
    }
  }
}

output {
  http {
    verb => "POST"
    url => "http://你的API地址"
    body_format => "json"
    body => {
      "instances" => "%{api_instances}"
    }
  }
}

关键注意点 #

• 字段引用：Logstash里用%{字段名}来引用事件中的字段，确保你的字段名和Kafka/ES里的实际字段一致
• HTTP插件配置：如果API返回响应，你可以在output的http插件里添加response_code => 200来判断请求是否成功，或者用tags标记失败的事件
• 语法检查：配置文件要严格遵循YAML语法，注意缩进，每个键值对的冒号后面要有空格

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