Logstash如何配置多输入流实现跨库关联查询输出

Logstash 内所有 input 块默认并行启动、独立运行，既不存在先后执行顺序，也不会自动在不同input插件之间共享传递变量。你写的配置里第二个JDBC查询启动时，根本拿不到第一个JDBC插件查出来的时间戳参数，自然无法按预期执行。

推荐用Logstash原生的pipeline串联+jdbc_streaming 插件实现，稳定性最高，适配所有不同类型数据库的场景：

方案配置步骤 #

1. 先修改Logstash目录下的config/pipelines.yml，定义两个存在依赖关系的pipeline，保证执行顺序：

- pipeline.id: fetch_base_timestamp
  path.config: "config/pipelines/fetch_ts.conf"
- pipeline.id: query_business_data
  path.config: "config/pipelines/query_b.conf"

2. 新建第一个pipeline配置文件config/pipelines/fetch_ts.conf，专门负责从数据库A拉取时间戳，拉到后将结果转发给下游pipeline：

input {
  jdbc {
    # 数据库A的连接信息
    jdbc_connection_string => "jdbc:你的数据库A连接串"
    jdbc_user => "数据库A账号"
    jdbc_password => "数据库A密码"
    jdbc_driver_class => "对应数据库的驱动类"
    # 按你的业务需求配置执行周期，比如每10分钟执行一次就写 "*/10 * * * *"
    schedule => "你的执行周期"
    statement => "SELECT 你要获取的时间戳字段 AS sync_timestamp FROM 对应的表 WHERE 你的筛选条件"
  }
}
output {
  pipeline {
    send_to => ["biz_query_worker"]
  }
}

3. 新建第二个pipeline配置文件config/pipelines/query_b.conf，接收上游传来的时间戳，用该参数查询数据库B，最后将结果写入Elasticsearch：

input {
  pipeline {
    address => biz_query_worker
  }
}
filter {
  # 用收到的时间戳作为参数查询数据库B
  jdbc_streaming {
    jdbc_connection_string => "jdbc:你的数据库B连接串"
    jdbc_user => "数据库B账号"
    jdbc_password => "数据库B密码"
    jdbc_driver_class => "对应数据库的驱动类"
    statement => "SELECT * FROM 你的业务表 WHERE 时间字段 >= :sync_ts"
    parameters => { "sync_ts" => "sync_timestamp" }
    # 查询结果暂存到临时字段
    target => "result_set"
  }
  # 多条查询结果拆分为独立事件
  split {
    field => "result_set"
  }
  # 把查询结果的字段提升到事件根层级，清理冗余字段
  ruby {
    code => "
      event.get('result_set').each { |k, v| event.set(k, v) }
      event.remove('result_set')
      event.remove('sync_timestamp')
      event.remove('@version')
    "
  }
}
output {
  elasticsearch {
    hosts => ["你的ES服务地址"]
    index => "目标索引名称"
    document_id => "%{业务表的主键字段}"
  }
}

其他可选方案 #

如果你的数据库A和数据库B类型相同、网络互通且支持跨库关联查询，也可以直接在单个JDBC input中写跨库关联SQL，一次查询拿到目标结果，配置更简单，但跨库查询性能较差，仅适合数据量小的场景。

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