使用logstash将mysql与elasticsearch同步

我有带有嵌套标签字段的产品文档，每次从数据库获取更新的产品记录时都使用logstash，但是当单独更新与产品相关的标签时，logstash如何能够检测到子记录已更新然后更新产品文档的问题是可能的，以及如何实现那个目标？

这是Logstash配置文件

input {
jdbc {
jdbc_driver_library => "mysql-connector-java-5.1.47-bin.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://localhost:3306/aurora_cms?useSSL=false&allowPublicKeyRetrieval=true"
jdbc_user => "root"
jdbc_password => "root"
statement => "select * from product where updated_date > :sql_last_value"
schedule=>"*/10 * * * * *"
}
}
filter {
jdbc_streaming {
jdbc_driver_library => "mysql-connector-java-5.1.47-bin.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://localhost:3306/aurora_cms?useSSL=false&allowPublicKeyRetrieval=true"
jdbc_user => "root"
jdbc_password => "root"
statement => "select name
from tag
where product_id= :id"
parameters => { id => "product_id"}
target => "tags"
}
}
output {
elasticsearch {
hosts => ["localhost:9200"]
document_id => "%{product_id}"
index => "products"
document_type => "products"
}
}