logstash的JDBC输入插件可以执行多个sql任务吗？

Question

现在有一个JDBC任务从db读取为

input {
  jdbc {
    jdbc_connection_string => "jdbc:mysql://${MYSQL_MAIN_HOST}/${MYSQL_DATABASE}"
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    jdbc_page_size => 10000
    jdbc_paging_enabled => true
    jdbc_password => "${MYSQL_PASSWORD}"
    jdbc_user => "${MYSQL_USER}"
    schedule => "0 1 * * *"
    statement_filepath => "/usr/share/logstash/pipeline/sql/select_posts.sql"
    tracking_column => "updated_at"
    tracking_column_type => "numeric"
    use_column_value => true
    last_run_metadata_path => "/usr/share/logstash/jdbc_last_run/select_posts_last_value"
  }
}

/usr/share/logstash/pipeline/sql/select_posts.sql

SELECT
  id,
  title,
  body
FROM posts

当body项转到非常大的数据时，此任务很重。所以我想在第一次搜索时删除它，因为：

SELECT
  id,
  title
FROM posts

然后获取ID并使用它们再次查找body。

SELECT
  body
FROM posts
WHERE id in (IDs)

设置为输出目标。甚至更新输出目标也是可以的。

那么，在这种情况下，logstash可以逐条语句读取语句吗？

Answer 1

如果您想使用logstash执行两个单独的sql查询，您可以这样做：

• query1.sql

SELECT
  id,
  title
FROM posts

• query2.sql

SELECT
  body
FROM posts
WHERE id in (IDs)

• logstash.conf

input {
  jdbc {
    jdbc_connection_string => "jdbc:mysql://${MYSQL_MAIN_HOST}/${MYSQL_DATABASE}"
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    jdbc_page_size => 10000
    jdbc_paging_enabled => true
    jdbc_password => "${MYSQL_PASSWORD}"
    jdbc_user => "${MYSQL_USER}"
    schedule => "0 1 * * *"
    statement_filepath => <PATH TO query1.sql>  -----query1.sql
    tracking_column => "updated_at"
    tracking_column_type => "numeric"
    use_column_value => true
    last_run_metadata_path => "/usr/share/logstash/jdbc_last_run/select_posts_last_value"
  }

 jdbc {
    jdbc_connection_string => "jdbc:mysql://${MYSQL_MAIN_HOST}/${MYSQL_DATABASE}"
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    jdbc_page_size => 10000
    jdbc_paging_enabled => true
    jdbc_password => "${MYSQL_PASSWORD}"
    jdbc_user => "${MYSQL_USER}"
    schedule => "0 1 * * *"
    statement_filepath => "<PATH TO query2.sql>"    -----query2.sql
    tracking_column => "updated_at"
    tracking_column_type => "numeric"
    use_column_value => true
    last_run_metadata_path => "/usr/share/logstash/jdbc_last_run/select_posts_last_value"
  }
}