logstash-input-jdbc增量、全量数据同步

曾经终败给现在 2022-04-13 05:36 240阅读 0赞

# 一、场景 #

笔者在mysql数据同步到ES中，发现第一次同步时需要全量的数据，之后则需要定时去同步增量数据，所以笔者提供增量和全量同步的conf供读者参考

--------------------

# 二、解决方案 #

## 1、全量数据同步 ##

具体如何执行可参考[https://blog.csdn.net/w\_linux/article/details/84555506][https_blog.csdn.net_w_linux_article_details_84555506]，这里提供conf的配置

input {
      jdbc {
        jdbc_driver_library => "./config/mysql-connector-java-5.1.39.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        # 数据库相关配置
        jdbc_connection_string => "jdbc:ip:port/jdcomm?characterEncoding=UTF-8&useSSL=false"
        jdbc_user => "root"
        jdbc_password => "password"
        statement => "SELECT * FROM blwjxb_z"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "50000"
        schedule => "*/10 * * * *"
      }
    }
    
    filter {
       json {
            source => "message"
            remove_field => ["message"]
        }
    }
    
    output {
      stdout {
        codec => rubydebug
      }
      elasticsearch {
        hosts => "localhost"
        #将mysql数据加入myindex索引下，会自动创建
        index => "myindex"
        # 自增ID 需要关联的数据库中有有一个id字段，对应索引的id号
        document_id => "%{id}"
      }        
    }

## 2、增量数据同步 ##

input {
      jdbc {
        jdbc_driver_library => "./config/mysql-connector-java-5.1.39.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        # 数据库相关配置
        jdbc_connection_string => "jdbc:ip:port/jdcomm?characterEncoding=UTF-8&useSSL=false"
        jdbc_user => "root"
        jdbc_password => "password"
        statement => "SELECT * FROM blwjxb_z where id > :sql_last_value"
        #使用其它字段追踪，而不是用时间
        use_column_value => true
        #追踪的字段
        tracking_column => id
        record_last_run => true
        #上一个sql_last_value值的存放文件路径, 必须要在文件中指定字段的初始值
        last_run_metadata_path => "./config/station_parameter.txt"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "50000"
        schedule => "* * * * *"
      }
    }
    
    filter {
       json {
            source => "message"
            remove_field => ["message"]
        }
    }
    
    output {
      stdout {
        codec => rubydebug
      }
      elasticsearch {
        hosts => "localhost"
        #将mysql数据加入myindex索引下，会自动创建
        index => "myindex"
        # 自增ID 需要关联的数据库中有有一个id字段，对应索引的id号
        document_id => "%{id}"
      }        
    }

思路就是每次脚本定时执行的时候会去找id>station\_parameter.txt中设置的数值，每次增量数据同步后，station\_parameter.txt中的数值会自动更新。

station\_parameter.txt中的数据初始如下图

![20181127173610704.png][]

[https_blog.csdn.net_w_linux_article_details_84555506]: https://blog.csdn.net/w_linux/article/details/84555506
[20181127173610704.png]: /images/20220413/6402d3b262524db19ef26f97db86710b.png