LogStash+MySQL+Elasticsearch实现数据增量导入

input {

  jdbc {

    # 连接地址

    jdbc_connection_string => "jdbc:mysql://192.168.1.2:3306/ego?useUnicode=true&characterEncoding=UTF-8&serverTimezOne=UTC"

    # 数据库用户名和密码

    jdbc_user => "root"

    jdbc_password => "root"

    # 驱动类，如果使用低版本的logstash，需要再增加配置 jdbc_driver_library，配置驱动包所在位置

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

    # 是否开启分页逻辑

    jdbc_paging_enabled => true

    # 分页的长度是多少

    jdbc_page_size => "2000"

    # 时区

    jdbc_default_timezOne=> "Asia/Shanghai"

    # 执行的SQL

    statement => "select id, title, sell_point, price, image, updated from tb_item where updated >= :sql_last_value order by updated asc"

    # 执行SQL的周期， [秒] 分钟 小时 天 月 年

    schedule => "* * * * *"

    # 是否使用字段的值作为比较策略

    use_column_value => true

    # 作为比较策略的字段名称

    tracking_column => "updated"

    # 作为比较策略的字段类型，可选为numberic和timestamp

    tracking_column_type => "timestamp"

    # 记录最近的比较策略字段值的文件是什么，相对寻址路径是logstash的安装路径

    last_run_metadata_path => "./ego-items-db2es-last-value"

    # 是否每次执行SQL的时候，都删除last_run_metadata_path文件内容

    clean_run => false

    # 是否强制把ES中的字段名都定义为小写。

    lowercase_column_names => false

  }

}

output {

  elasticsearch {

    hosts => ["http://192.168.89.140:9200", "http://192.168.89.141:9200"]

    index => "ego-items-index"

    action => "index"

    document_id => "%{id}"

  }

}