我已将 logstash.conf 配置为动态插入我的数据库的数据，但问题是:
当我更改表中的一行时，该行未在我的索引中更新，因为我只在 sql_last_value 之后插入新值，我认为我们是一个触发器，但我不确定我该怎么做。
input {
jdbc {
jdbc_connection_string =>"jdbc:mysql://localhost:3306/blog"
jdbc_user =>"root"
jdbc_password =>""
jdbc_driver_library =>"C:\Users\saidb\Downloads\mysql-connector-java-5.1.47\mysql-connector-java-5.1.47.jar"
jdbc_driver_class =>"com.mysql.jdbc.Driver"
schedule =>"* * * * *"
statement =>"SELECT * FROM blog_pro WHERE id >:sql_last_value"
use_column_value =>true
tracking_column =>id
}
}
output {
elasticsearch {
hosts =>"localhost:9200"
index =>"blog_pro"
document_type =>"data"
}
}
最佳答案
如果您使用 id 来选择行，则不能这样做。
您有 2 个选择，
每次选择所有行并使用查询
SELECT * FROM blog_pro将它们发送到 ES，根据您的情况我认为这不是一个好的做法。创建一个新列
last_modified_time，其中包含记录(行)的最后修改时间戳。然后用它来过滤行。注意属性tracking_column_type => "timestamp"
语句 =>“SELECT * FROM blog_pro WHERE last_modiefied_time >:sql_last_value”
use_column_value =>true
tracking_column =>last_modified_time
tracking_column_type => "时间戳"
这是完整的 logstash 配置
input {
jdbc {
jdbc_connection_string =>"jdbc:mysql://192.168.3.57:3306/blog_pro"
jdbc_user =>"dush"
jdbc_password =>"dush"
jdbc_driver_library =>"F:\logstash-6.2.2\bin\mysql-connector-java-5.1.6.jar"
jdbc_driver_class =>"com.mysql.jdbc.Driver"
schedule =>"* * * * *"
statement =>"SELECT * FROM blog_pro WHERE last_modified_time >:sql_last_value"
use_column_value =>true
tracking_column =>last_modified_time
tracking_column_type => "timestamp"
}
}
output
{
#output to elasticsearch
elasticsearch {
hosts => [ "192.168.1.245:9201" ]
action=>update
# "%{id}" - > primary key of the table
document_id => "%{id}"
doc_as_upsert =>true
}
}
请注意，您可能需要清除索引并使用此配置开始索引。我对此进行了测试并且工作正常。
Elasticsearch 版本 = 5.x.x
logstash 版本=6.2.2
关于mysql - 我如何使用 logstash 从 MySQL 动态更新我的数据(不重复)？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/54421191/