这是因为每次执行cron作业时，查询都会获取所有数据。另外，您没有在elasticsearch输出中提供自定义id，因此它会为每个文档创建动态ID，因此索引中会有更多数据（具有不同唯一ID的重复数据）。
您可以使用存储上次搜索日期的sql_last_value参数，并使用created_date或updated_date上的where条件更新查询。这将第一次从数据库中获取所有数据，第二次以后仅获取新创建或更新的数据。

input {
  jdbc {
    jdbc_driver_library => "/correct_path/java/mysql-connector-java-8.0.27.jar" 
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/my_db"
    jdbc_user => "user" 
    jdbc_password => "password" 
    jdbc_paging_enabled => true
    schedule => "*/5 * * * * *"
    statement => 'select * from my_table where created_date > :sql_last_value or updated_date > :sql_last_value'
  }
}

output {
    elasticsearch {
      user => "test"
      password => "test"
      hosts => ["localhost:9200"] 
      index => "my_index"
    }
    stdout { codec => "rubydebug" }
}

PS：我不是SQL专业人士，所以我的查询可能会有问题。但我希望你能明白这个想法。