如何在Logstash的Beats与JDBC输入插件间传递数据？

Logstash的不同输入插件是独立运行的，彼此之间没有直接传递数据的机制——你没法从beats输入里拿值直接传给另一个jdbc输入的查询语句，因为两个输入是并行的数据流，互相不感知。

要实现用Filebeat日志中的value值作为JDBC查询条件的需求，正确的做法是把JDBC查询逻辑放到filter阶段，使用jdbc_streaming或jdbc_static插件，这样就能在处理Filebeat事件的过程中，动态引用事件里的字段值执行查询。

示例配置 #

input {
  beats {
    port => 5044 # 监听Filebeat的端口
    # 这里可以添加grok等解析规则，把日志中的value提取成字段，比如：
    # add_field => { "value" => "%{your_pattern_here}" }
  }
}

filter {
  # 假设已经通过前面的解析（比如grok）得到了名为`value`的字段
  jdbc_streaming {
    jdbc_driver_library => "/path/to/mysql-connector-java.jar"
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/your_db"
    jdbc_user => "your_user"
    jdbc_password => "your_password"
    # 用参数化查询避免SQL注入，?会被parameters里的value替换
    statement => "SELECT * from table WHERE ruslit = ?"
    parameters => { "value" => "%{value}" }
    # 查询结果会放到这个字段里
    target => "jdbc_query_results"
    # 可选：设置缓存，避免重复查询相同值
    use_cache => true
    cache_size => 1000
  }
}

output {
  stdout { codec => rubydebug }
  # 这里可以添加其他输出，比如elasticsearch
}

关键说明 #

• 输入仅保留beats：JDBC不再作为输入，而是作为filter环节的一部分，随Filebeat事件触发查询
• 参数化查询：用?占位符+parameters配置的方式传递字段值，比直接拼接字符串更安全，能防止SQL注入
• 字段解析前提：要确保Filebeat过来的日志已经被正确解析，提取出了value字段（可以用grok、dissect等filter插件完成解析）

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