logstash-input-jdbc把数据从mysql导入es中，sql_last_value问题

  今天测试使用Logstash-input-jdbc同步mysql数据，追踪create_date_字段作为sql_last_valued的值，Logstash配置如下：

input {

    jdbc {

                jdbc_driver_library => "/home/software/logstash-6.4.3/lib/mysql-connector-java-5.1.41-bin.jar"

                jdbc_driver_class => "com.mysql.jdbc.Driver"

                jdbc_connection_string => "jdbc:mysql://192.168.83.128:3306/elk"

                jdbc_user => "root"

                jdbc_password => "Root_123"

                schedule => "* * * * *"

                clean_run => false

                use_column_value => true

                tracking_column => "create_date_"

                tracking_column_type => timestamp

                record_last_run => true

                last_run_metadata_path => "/home/software/logstash-6.4.3/local_email_last_run.txt"

                jdbc_page_size => 5000

                jdbc_paging_enabled => "true"

                lowercase_column_names => false

                statement => "SELECT * from ftsp_infra_send_email where create_date_ > :sql_last_value"

                type => "local-email"

        }

}

output {

        if [type] == "local-email" {

                elasticsearch {

                        hosts  => ["192.168.83.128:9200","192.168.83.129:9200"]

                        index  => "%{type}-%{+YYYY.MM.dd}"

                }

        }

}

    Logstash正常启动开始同步数据，几分钟local_email_last_run.txt文件生成时间戳，但这个时间戳不是最新一条记录的create_date_，是前面某条记录的时间。导致几分钟后重新同步，多条数据重复写入elasticsearch。Logstash多次读取，终于把时间戳更新到mysql最新一条记录的时间，后续增量读取就正常了。
    可是有多条数据被重复读了，1000条原始数据，最后elasticsearch中生成了1062条，请问怎么解决这个重复读的问题？能否第一次就直接读完然后将sql_last_value更新至最新一条记录的create_date_的时间呢？
    日志记录截图如下，sql_last_value的值经过多次更新。