logstash定时同步问题

Question

用logstash从MySQL中同步数据到es，在配置文件中，配置的有定时执行的任务，但是只有在当天这个定时任务会执行，后面这个定时任务就不会执行了，配置文件如下

input {
        stdin {}
        jdbc {
            #===================这是你需要修改的第1个地方===============#

                #数据库连接地址
                jdbc_connection_string => "jdbc:mysql://192.168.111.132:3306/es?characterEncoding=UTF-8&autoReconnect=true"
                jdbc_user => "root"
                jdbc_password => "qwer1234"
                # MySQL依赖包路径；
                jdbc_driver_library => "/opt/module/logstash-6.8.6/mysql-connector-java-5.1.49.jar"
                #查询数据库语句（查询出的字段名要和es索引的字段名相同）
                statement => "SELECT * FROM t1 where id >=1 order by id asc limit 1000"

                # 需要记录的字段，用于增量同步，需是数据库字段
                tracking_column => "createtime"
                # record_last_run上次数据存放位置；
                last_run_metadata_path => "/opt/module/logstash-6.8.6/data/last_time.txt"

                #====================第1个地方修改结束=====================#

                type => "jdbc"
                # the name of the driver class for mysql
                jdbc_driver_class => "com.mysql.jdbc.Driver"
                # 数据库重连尝试次数
                connection_retry_attempts => "3"
                 # 判断数据库连接是否可用，默认false不开启
                jdbc_validate_connection => "true"
                 # 数据库连接可用校验超时时间，默认3600S
                jdbc_validation_timeout => "3600"
                 # 开启分页查询（默认false不开启）；
                jdbc_paging_enabled => "true"
                 # 单次分页查询条数（默认100000,若字段较多且更新频率较高，建议调低此值）；
                jdbc_page_size => "500"
                 # statement为查询数据sql，如果sql较复杂，建议配通过statement_filepath配置sql文件的存放路径；
                 # sql_last_value为内置的变量，存放上次查询结果中最后一条数据tracking_column的值，此处即为ModifyTime；
                 # statement_filepath => "mysql/jdbc.sql"
                 # 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）；
                lowercase_column_names => false
                 # Value can be any of: fatal,error,warn,info,debug，默认info；
                sql_log_level => warn
                 #
                 # 是否记录上次执行结果，true表示会将上次执行结果的tracking_column字段的值保存到last_run_metadata_path指定的文件中；
                record_last_run => true
                 # 需要记录查询结果某字段的值时，此字段为true，否则默认tracking_column为timestamp的值；
                use_column_value => true
                 # Value can be any of: numeric,timestamp，Default value is "numeric"
                tracking_column_type => timestamp
                 # 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false；
                clean_run => false
                 #
                 # 同步频率(分 时 天 月 年)，默认每分钟同步一次；
                 schedule => "0 10 * * *"               
        }
}

output {
        elasticsearch {
             #===================这是你需要修改的第2个地方===============#

                 # host => "localhost"
                 # port => "9200"
                 # 配置ES集群地址
         hosts => ["192.168.111.132:9200"]
         user => "sxga"
         password => "sxga2024"
         # 启用SSL/TLS加密（前提是开启了HTTPS访问）
         ssl => false
         # 禁用证书验证（前提是开启了HTTPS访问）
         ssl_certificate_verification => false
         # 证书地址（如果启用SSL/TLS加密的话，即使禁用证书验证，也需要配置否则会报错）
         #cacert => "/Users/chencunyou/Desktop/java/logstash-7.10.2/CloudSearchService.cer"
                 # 你的索引名字，必须小写
                 index => "sgass"
                 # 数据唯一索引（建议使用数据库表唯一索引id）
                 document_id => "%{id}"

                 document_type => "_doc"

         #====================第2个地方修改结束=====================#
        }
        stdout {
                codec => json_lines
        }
}

Answer 1

修改后的配置如下：（如若有用，还请题主采纳！）

input {
    jdbc {
        jdbc_connection_string => "jdbc:mysql://192.168.111.132:3306/es?characterEncoding=UTF-8&autoReconnect=true"
        jdbc_user => "root"
        jdbc_password => "qwer1234"
        jdbc_driver_library => "/opt/module/logstash-6.8.6/mysql-connector-java-5.1.49.jar"
        statement => "SELECT * FROM t1 WHERE id >= 1 ORDER BY id ASC LIMIT 1000"
        tracking_column => "createtime"
        last_run_metadata_path => "/opt/module/logstash-6.8.6/data/last_time.txt"
        record_last_run => true
        use_column_value => true
        tracking_column_type => timestamp
        clean_run => false
        schedule => "* * * * *"  # 每分钟执行一次
    }
}

output {
    elasticsearch {
        hosts => ["192.168.111.132:9200"]
        user => "sxga"
        password => "sxga2024"
        index => "sgass"
        document_id => "%{id}"
        document_type => "_doc"
    }
    stdout {
        codec => json_lines
    }
}