mysql - 查询执行完成后自动更改 logstash jdbc 查询

Question

我正在尝试使用 logstash jdbc 输入将我们所有的订单数据导入 elasticsearch。 sql查询涉及跨7个表的连接，连接后不可能建立索引，表之间存在一对多关系。加入后大约有 3M 行。

我想逐日导入订单数据，因为记录会更少，而不是针对迄今为止的完整订单运行整个查询并对它们进行分页。如果我查询整个数据，数据库会自行计算记录数。

如何在不停止 logstash 和动态配置 SQL 日期参数的情况下日复一日地运行导入。

这是我当前的 logstash 作业配置

input {
jdbc { 
jdbc_connection_string => "jdbc:mysql://dbhostname:3306/mydatabase?zeroDateTimeBehavior=CONVERT_TO_NULL"
jdbc_user => "username"
jdbc_password => "password"
jdbc_driver_library => "/home/username/mysql-connector-java-8.0.11.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
statement => "select * from table1 t1, table2 t2 ,table3 t3 , table4 t4, table5 t5, table6 t6 where some_condition and some_other_condition and t1.created_at between 'date1_start' and 'date1_end'"
}
}
output {
stdout { codec => json_lines }
elasticsearch {
"hosts" => "localhost:9200"
"index" => "orderdata_prod"
"document_type" => "data"
}
}

我想用新的日期值动态更改配置中提供的语句。我怎样才能做到这一点？或者是否有其他方法可以导入此类数据？

Answer 1

我猜您可能担心生产数据库可能会崩溃，但您可以使用 logstash 的分页将单个查询分解为多个较小的查询来降低风险。但是您应该一次性运行导入，因为不能保证查询之间的顺序。

如果你想限制每个查询下载 100000 行，那么在你的 logstsh conf 文件中添加这两个参数:

jdbc_paging_enabled => true
jdbc_page_size => 100000

您也可以通过将 fetch size 参数降低到 db 的默认值来尝试，但请注意，此参数只是 jdbc 的一个提示，有时可能不起作用。例如，Oracle 将默认提取大小设置为 10，因此您可以将其降低到比方说 5:

jdbc_fetch_size => 5