logstash

jdbc.conf

启动命令

./logstash -f /newdata/es/logstash-8.11.1/jdbc.conf

input {
    jdbc {
        jdbc_connection_string => "jdbc:mysql://10.1.2.5:336/jeecg-boot?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=false&serverTimezone=GMT%2B8"
        jdbc_user => "root"
        jdbc_password => "sh3433333"
        jdbc_driver_library => "D:\\devsoft\\logstash-8.13.4\\mysql-connector-j-8.0.33.jar"
        jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "10000"
        statement_filepath => "msg.sql"
        use_column_value => true
        tracking_column => "create_time"
        tracking_column_type => "timestamp"
        record_last_run => true
        last_run_metadata_path => "msg-id.last"
        schedule => "*/5 * * * *"
        type => "msg"
    }
}

filter {
    #json {
        #source => "message"
        #remove\_field => ["message"]
    #}
    ruby {
        code => "
            if event.get('create_time')
                # 先将 create_time 转换为字符串
                create_time_str = event.get('create_time').to_s
                # 然后截取字符串的前7位
                event.set('index_suffix', create_time_str[0,7])
            end
        "
    }


}

output {
    elasticsearch {
        hosts => "localhost:9200"
        index => "alarm_indicator_recording-%{index_suffix}"
        document_type => "%{type}"
    }
    stdout {
        codec => json_lines
    }
}

msg.sql

select 
id, sn_num, num, indicator_name, indicator_type, indicator_factor, indicator_num, quality, alarm_level, create_by, create_time, update_by, update_time, del_flag, alarm_location, description 
from 
    alarm_indicator_recording
where 
    create_time > date_add(:sql_last_value,INTERVAL 8 HOUR)

msg-id.last

--- 2024-03-21 03:16:23.000000000 Z