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

Question

我正在尝试使用 logstash jdbc 输入将我们所有的订单数据导入 elasticsearch。 sql查询涉及跨7个表的join，join后不能建立索引，表之间是一对多的关系。加入后大约有 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 的分页将单个查询分解为多个较小的查询来降低风险。但是您应该一次性运行导入，因为不能保证查询之间的顺序。

如果您想将每个查询限制为下载 100 000 行，请在您的 logstsh conf 文件中添加这两个参数：

jdbc_paging_enabled => true
jdbc_page_size => 100000

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

jdbc_fetch_size => 5